aboutsummaryrefslogtreecommitdiff
path: root/syncapi/storage/sqlite3/stream_id_table.go
diff options
context:
space:
mode:
Diffstat (limited to 'syncapi/storage/sqlite3/stream_id_table.go')
-rw-r--r--syncapi/storage/sqlite3/stream_id_table.go12
1 files changed, 12 insertions, 0 deletions
diff --git a/syncapi/storage/sqlite3/stream_id_table.go b/syncapi/storage/sqlite3/stream_id_table.go
index e6bdc4fc..f73be422 100644
--- a/syncapi/storage/sqlite3/stream_id_table.go
+++ b/syncapi/storage/sqlite3/stream_id_table.go
@@ -18,6 +18,8 @@ CREATE TABLE IF NOT EXISTS syncapi_stream_id (
);
INSERT INTO syncapi_stream_id (stream_name, stream_id) VALUES ("global", 0)
ON CONFLICT DO NOTHING;
+INSERT INTO syncapi_stream_id (stream_name, stream_id) VALUES ("receipt", 0)
+ ON CONFLICT DO NOTHING;
`
const increaseStreamIDStmt = "" +
@@ -56,3 +58,13 @@ func (s *streamIDStatements) nextStreamID(ctx context.Context, txn *sql.Tx) (pos
err = selectStmt.QueryRowContext(ctx, "global").Scan(&pos)
return
}
+
+func (s *streamIDStatements) nextReceiptID(ctx context.Context, txn *sql.Tx) (pos types.StreamPosition, err error) {
+ increaseStmt := sqlutil.TxStmt(txn, s.increaseStreamIDStmt)
+ selectStmt := sqlutil.TxStmt(txn, s.selectStreamIDStmt)
+ if _, err = increaseStmt.ExecContext(ctx, "receipt"); err != nil {
+ return
+ }
+ err = selectStmt.QueryRowContext(ctx, "receipt").Scan(&pos)
+ return
+}