Skip to content

Commit

Permalink
Merge 3d4f611 into 9cd8c73
Browse files Browse the repository at this point in the history
  • Loading branch information
ABresting authored Oct 24, 2023
2 parents 9cd8c73 + 3d4f611 commit ea16d69
Show file tree
Hide file tree
Showing 10 changed files with 16 additions and 67 deletions.
2 changes: 1 addition & 1 deletion tests/waku_archive/test_driver_postgres.nim
Original file line number Diff line number Diff line change
Expand Up @@ -87,7 +87,7 @@ suite "Postgres driver":
require:
storedMsg.len == 1
storedMsg.all do (item: auto) -> bool:
let (pubsubTopic, actualMsg, digest, messageHash, storeTimestamp) = item
let (pubsubTopic, actualMsg, digest, storeTimestamp) = item
actualMsg.contentTopic == contentTopic and
pubsubTopic == DefaultPubsubTopic and
toHex(computedDigest.data) == toHex(digest) and
Expand Down
2 changes: 1 addition & 1 deletion tests/waku_archive/test_driver_sqlite.nim
Original file line number Diff line number Diff line change
Expand Up @@ -60,7 +60,7 @@ suite "SQLite driver":
check:
storedMsg.len == 1
storedMsg.all do (item: auto) -> bool:
let (pubsubTopic, msg, digest, messageHash, storeTimestamp) = item
let (pubsubTopic, msg, digest, storeTimestamp) = item
msg.contentTopic == contentTopic and
pubsubTopic == DefaultPubsubTopic

Expand Down
44 changes: 0 additions & 44 deletions tests/waku_archive/test_driver_sqlite_query.nim
Original file line number Diff line number Diff line change
Expand Up @@ -423,50 +423,6 @@ suite "SQLite driver - query by pubsub topic":

## Cleanup
(await driver.close()).expect("driver to close")

asyncTest "pubSubTopic messageHash match":
## Given
const pubsubTopic1 = "test-pubsub-topic1"
const pubsubTopic2 = "test-pubsub-topic2"
# take 2 variables to hold the message hashes
var msgHash1: seq[byte]
var msgHash2: seq[byte]

let driver = newTestSqliteDriver()
var putFutures = newSeq[Future[ArchiveDriverResult[void]]]()

let msg1 = fakeWakuMessage(contentTopic=DefaultContentTopic, ts=Timestamp(1))
putFutures.add(driver.put(pubsubTopic1, msg1, computeDigest(msg1, pubsubTopic1), msg1.timestamp))

let msg2 = fakeWakuMessage(contentTopic=DefaultContentTopic, ts=Timestamp(2))
putFutures.add(driver.put(pubsubTopic2, msg2, computeDigest(msg2, pubsubTopic2), msg2.timestamp))

discard waitFor allFinished(putFutures)

# get the messages from the database
let storedMsg = (waitFor driver.getAllMessages()).tryGet()

check:
# there needs to be two messages
storedMsg.len > 0
storedMsg.len == 2

# get the individual messages and message hash values
@[storedMsg[0]].all do (item1: auto) -> bool:
let (gotPubsubTopic1, gotMsg1, digest1, messageHash1, timestamp1) = item1
msgHash1 = messageHash1
true

@[storedMsg[1]].all do (item2: auto) -> bool:
let (gotPubsubTopic2, gotMsg2, digest2, messageHash2, timestamp2) = item2
msgHash2 = messageHash2
true

# compare of the messge hashes, given the context, they should be different
msgHash1 != msgHash2

## Cleanup
(await driver.close()).expect("driver to close")


suite "SQLite driver - query by cursor":
Expand Down
2 changes: 1 addition & 1 deletion tests/waku_archive/test_retention_policy.nim
Original file line number Diff line number Diff line change
Expand Up @@ -147,7 +147,7 @@ suite "Waku Archive - Retention policy":
check:
storedMsg.len == capacity
storedMsg.all do (item: auto) -> bool:
let (pubsubTopic, msg, digest, messageHash, storeTimestamp) = item
let (pubsubTopic, msg, digest, storeTimestamp) = item
msg.contentTopic == contentTopic and
pubsubTopic == DefaultPubsubTopic

Expand Down
2 changes: 1 addition & 1 deletion waku/waku_archive/archive.nim
Original file line number Diff line number Diff line change
Expand Up @@ -163,7 +163,7 @@ proc findMessages*(w: WakuArchive, query: ArchiveQuery): Future[ArchiveResult] {
## Build last message cursor
## The cursor is built from the last message INCLUDED in the response
## (i.e. the second last message in the rows list)
let (pubsubTopic, message, digest, messageHash, storeTimestamp) = rows[^2]
let (pubsubTopic, message, digest, storeTimestamp) = rows[^2]

# TODO: Improve coherence of MessageDigest type
let messageDigest = block:
Expand Down
2 changes: 1 addition & 1 deletion waku/waku_archive/driver.nim
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,7 @@ type
ArchiveDriver* = ref object of RootObj
OnErrHandler* = proc(errMsg: string) {.gcsafe, closure.}

type ArchiveRow* = (PubsubTopic, WakuMessage, seq[byte], seq[byte], Timestamp)
type ArchiveRow* = (PubsubTopic, WakuMessage, seq[byte], Timestamp)

# ArchiveDriver interface

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -138,7 +138,6 @@ proc toArchiveRow(r: Row): ArchiveDriverResult[ArchiveRow] =
return ok((pubSubTopic,
wakuMessage,
@(digest.toOpenArrayByte(0, digest.high)),
@(digest.toOpenArrayByte(0, digest.high)),
storedAt))

method getAllMessages*(s: PostgresDriver):
Expand Down
2 changes: 1 addition & 1 deletion waku/waku_archive/driver/queue_driver/queue_driver.nim
Original file line number Diff line number Diff line change
Expand Up @@ -138,7 +138,7 @@ proc getPage(driver: QueueDriver,

numberOfItems += 1

outSeq.add((key.pubsubTopic, data.msg, @(key.digest.data), @(key.digest.data), key.receiverTime))
outSeq.add((key.pubsubTopic, data.msg, @(key.digest.data), key.receiverTime))

currentEntry = if forward: w.next()
else: w.prev()
Expand Down
25 changes: 10 additions & 15 deletions waku/waku_archive/driver/sqlite_driver/queries.nim
Original file line number Diff line number Diff line change
Expand Up @@ -71,9 +71,8 @@ proc createTableQuery(table: string): SqlQueryStr =
" version INTEGER NOT NULL," &
" timestamp INTEGER NOT NULL," &
" id BLOB," &
" messageHash BLOB NOT NULL,"&
" storedAt INTEGER NOT NULL," &
" CONSTRAINT messageIndex PRIMARY KEY (storedAt, messageHash)" &
" CONSTRAINT messageIndex PRIMARY KEY (storedAt, id, pubsubTopic)" &
") WITHOUT ROWID;"

proc createTable*(db: SqliteDatabase): DatabaseResult[void] =
Expand Down Expand Up @@ -103,11 +102,11 @@ proc createHistoryQueryIndex*(db: SqliteDatabase): DatabaseResult[void] =


## Insert message
type InsertMessageParams* = (seq[byte], seq[byte], Timestamp, seq[byte], seq[byte], seq[byte], int64, Timestamp)
type InsertMessageParams* = (seq[byte], Timestamp, seq[byte], seq[byte], seq[byte], int64, Timestamp)

proc insertMessageQuery(table: string): SqlQueryStr =
"INSERT INTO " & table & "(id, messageHash, storedAt, contentTopic, payload, pubsubTopic, version, timestamp)" &
" VALUES (?, ?, ?, ?, ?, ?, ?, ?);"
"INSERT INTO " & table & "(id, storedAt, contentTopic, payload, pubsubTopic, version, timestamp)" &
" VALUES (?, ?, ?, ?, ?, ?, ?);"

proc prepareInsertMessageStmt*(db: SqliteDatabase): SqliteStmt[InsertMessageParams, void] =
let query = insertMessageQuery(DbTable)
Expand Down Expand Up @@ -198,26 +197,24 @@ proc deleteOldestMessagesNotWithinLimit*(db: SqliteDatabase, limit: int):
## Select all messages

proc selectAllMessagesQuery(table: string): SqlQueryStr =
"SELECT storedAt, contentTopic, payload, pubsubTopic, version, timestamp, id, messageHash" &
"SELECT storedAt, contentTopic, payload, pubsubTopic, version, timestamp, id" &
" FROM " & table &
" ORDER BY storedAt ASC"

proc selectAllMessages*(db: SqliteDatabase): DatabaseResult[seq[(PubsubTopic,
WakuMessage,
seq[byte],
seq[byte],
Timestamp)]] =
## Retrieve all messages from the store.
var rows: seq[(PubsubTopic, WakuMessage, seq[byte], seq[byte], Timestamp)]
var rows: seq[(PubsubTopic, WakuMessage, seq[byte], Timestamp)]
proc queryRowCallback(s: ptr sqlite3_stmt) =
let
pubsubTopic = queryRowPubsubTopicCallback(s, pubsubTopicCol=3)
wakuMessage = queryRowWakuMessageCallback(s, contentTopicCol=1, payloadCol=2, versionCol=4, senderTimestampCol=5)
digest = queryRowDigestCallback(s, digestCol=6)
messageHash = queryRowDigestCallback(s, digestCol=7)
storedAt = queryRowReceiverTimestampCallback(s, storedAtCol=0)

rows.add((pubsubTopic, wakuMessage, digest, messageHash, storedAt))
rows.add((pubsubTopic, wakuMessage, digest, storedAt))

let query = selectAllMessagesQuery(DbTable)
let res = db.query(query, queryRowCallback)
Expand Down Expand Up @@ -283,7 +280,7 @@ proc selectMessagesWithLimitQuery(table: string, where: Option[string], limit: u

var query: string

query = "SELECT storedAt, contentTopic, payload, pubsubTopic, version, timestamp, id, messageHash"
query = "SELECT storedAt, contentTopic, payload, pubsubTopic, version, timestamp, id"
query &= " FROM " & table

if where.isSome():
Expand Down Expand Up @@ -364,20 +361,18 @@ proc selectMessagesByHistoryQueryWithLimit*(db: SqliteDatabase,
DatabaseResult[seq[(PubsubTopic,
WakuMessage,
seq[byte],
seq[byte],
Timestamp)]] =


var messages: seq[(PubsubTopic, WakuMessage, seq[byte], seq[byte], Timestamp)] = @[]
var messages: seq[(PubsubTopic, WakuMessage, seq[byte], Timestamp)] = @[]
proc queryRowCallback(s: ptr sqlite3_stmt) =
let
pubsubTopic = queryRowPubsubTopicCallback(s, pubsubTopicCol=3)
message = queryRowWakuMessageCallback(s, contentTopicCol=1, payloadCol=2, versionCol=4, senderTimestampCol=5)
digest = queryRowDigestCallback(s, digestCol=6)
messageHash = queryRowDigestCallback(s, digestCol=7)
storedAt = queryRowReceiverTimestampCallback(s, storedAtCol=0)

messages.add((pubsubTopic, message, digest, messageHash, storedAt))
messages.add((pubsubTopic, message, digest, storedAt))

let query = block:
let where = whereClause(cursor, pubsubTopic, contentTopic, startTime, endTime, ascending)
Expand Down
1 change: 0 additions & 1 deletion waku/waku_archive/driver/sqlite_driver/sqlite_driver.nim
Original file line number Diff line number Diff line change
Expand Up @@ -66,7 +66,6 @@ method put*(s: SqliteDriver,
## Inserts a message into the store
let res = s.insertStmt.exec((
@(digest.data), # id
@(digest.data), # messageHash
receivedTime, # storedAt
toBytes(message.contentTopic), # contentTopic
message.payload, # payload
Expand Down

0 comments on commit ea16d69

Please sign in to comment.