From eb5b286476cc12d6ec5c11c1df467f036cd4afed Mon Sep 17 00:00:00 2001 From: Jose Olarte III Date: Thu, 27 Aug 2026 20:20:51 +0800 Subject: [PATCH] Persist independent Endorser and Partner alertSearch cursors per user so later daily runs can resume after the last complete retrieval. Advance each cursor only on success, leave empty/pagination/error results unchanged, and keep this unwired from the scheduler. --- CHANGELOG.md | 5 + README.md | 10 +- package.json | 2 +- src/alertSearch/cursors.test.ts | 385 +++++++++++++++++++++++++++++ src/alertSearch/cursors.ts | 137 ++++++++++ src/alertSearch/index.ts | 7 + src/alertSearch/retrieve.ts | 7 +- src/db/alertSearchCursorsSqlite.ts | 93 +++++++ src/db/sqlite.ts | 17 +- 9 files changed, 656 insertions(+), 7 deletions(-) create mode 100644 src/alertSearch/cursors.test.ts create mode 100644 src/alertSearch/cursors.ts create mode 100644 src/db/alertSearchCursorsSqlite.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index 32a9a8c..2145b21 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,11 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [0.1.7] - 2026.08.27 +### Added +- Persistent per-user Endorser and Partner alertSearch cursors (advanced only after complete retrievals) + + ## [0.1.6] - 2026.08.27 ### Added - AlertSearch retrieval layer for Endorser and Partner GET endpoints (pagination, independent cursors; not scheduled yet) diff --git a/README.md b/README.md index e60035a..f32853c 100644 --- a/README.md +++ b/README.md @@ -71,7 +71,9 @@ Authorization: Bearer - `{ENDORSER_URL}/api/v2/report/alertSearch` - `{PARTNER_URL}/api/partner/alertSearch` -The **delegated** JWT is sent as `Authorization: Bearer`. Pass independent `endorserAfterId` / `partnerAfterId` ULIDs, or omit them on first run. Nearby search uses the alertSearch bbox (`minLocLat`, `maxLocLat`, `minLocLon`, `maxLocLon`). +The **delegated** JWT is sent as `Authorization: Bearer`. Pass independent `endorserAfterId` / `partnerAfterDate` (or omit them on first run). Nearby search uses the alertSearch bbox (`minLocLat`, `maxLocLat`, `minLocLon`, `maxLocLon`). + +`loadAlertSearchCursors` / `advanceAlertSearchCursors` persist those bounds per user DID in SQLite. Cursors advance only after a complete `success` retrieval (not `empty`, `pagination`, or errors). They are not wired to the scheduler yet. ## Storage @@ -102,6 +104,12 @@ Tables `alert_authorization_batches` and `alert_authorization_jwts` hold a user' Unique on `(batch_pk, sequence)` and on `(user_id, day)` for unused rows. Indexes also exist on `(user_id, status)`, `(user_id, day)`, and `batch_pk`. +Table `alert_search_cursors` holds one row per user DID: + +- `endorser_after_id` — last complete Endorser ULID (`afterId`), or null +- `partner_after_at` — last complete Partner `updatedAt` bound (`afterDate`), or null +- `created_at`, `updated_at` + The schema is created automatically on startup if the database or tables do not already exist. New tables are added with `CREATE TABLE IF NOT EXISTS`; existing `fcm_registrations` rows are not migrated or altered. ### JSON → SQLite diff --git a/package.json b/package.json index 422ac5d..d03f2cf 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "notification-wakeup-service", - "version": "0.1.6", + "version": "0.1.7", "private": true, "type": "module", "packageManager": "pnpm@11.4.0", diff --git a/src/alertSearch/cursors.test.ts b/src/alertSearch/cursors.test.ts new file mode 100644 index 0000000..a42a8ad --- /dev/null +++ b/src/alertSearch/cursors.test.ts @@ -0,0 +1,385 @@ +import assert from "node:assert/strict"; +import { mkdtemp, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import path from "node:path"; +import { afterEach, beforeEach, describe, it } from "node:test"; +import { alertSearchCursorsDb } from "../db/alertSearchCursorsSqlite.js"; +import { closeDatabase } from "../db/sqlite.js"; +import { + advanceAlertSearchCursors, + loadAlertSearchCursors, + maxEndorserAlertSearchUlid, +} from "./cursors.js"; +import { retrieveAlertSearch } from "./retrieve.js"; +import type { AlertSearchSourceResult, RetrieveAlertSearchResult } from "./retrieve.js"; +import { ALERT_SEARCH_PAGE_SIZE } from "./types.js"; +import type { EndorserAlertSearchData, PartnerAlertSearchData } from "./types.js"; + +const USER = "did:ethr:0xcursoruser"; +const OTHER_USER = "did:ethr:0xother"; +const JWT = "delegated.jwt.token"; + +function ulid(n: number): string { + return `01H${String(n).padStart(23, "0")}`; +} + +function emptyEndorser(): EndorserAlertSearchData { + return { + claims: [], + personalPlanContributions: [], + trackedPlanUpdates: [], + trackedPlanClaims: [], + plansNearby: [], + }; +} + +function emptyPartner(): PartnerAlertSearchData { + return { profilesNearby: [] }; +} + +function endorserResult( + outcome: AlertSearchSourceResult["outcome"], + data: EndorserAlertSearchData = emptyEndorser() +): AlertSearchSourceResult { + return { outcome, pageCount: 1, data }; +} + +function partnerResult( + outcome: AlertSearchSourceResult["outcome"], + data: PartnerAlertSearchData = emptyPartner() +): AlertSearchSourceResult { + return { outcome, pageCount: 1, data }; +} + +function combined( + endorser: AlertSearchSourceResult, + partner: AlertSearchSourceResult +): RetrieveAlertSearchResult { + return { + data: { ...endorser.data, ...partner.data }, + empty: false, + endorser, + partner, + }; +} + +function jsonResponse(body: unknown, status = 200): Response { + return new Response(JSON.stringify(body), { + status, + headers: { "Content-Type": "application/json" }, + }); +} + +function emptyEndorserBody() { + return { + data: { + claims: [], + personalPlanContributions: [], + trackedPlanUpdates: [], + trackedPlanClaims: [], + plansNearby: [], + }, + }; +} + +function emptyPartnerBody(overrides?: Record) { + return { data: { profilesNearby: [], ...overrides } }; +} + +describe("alertSearch cursor persistence", () => { + let dir: string; + let previousDataDir: string | undefined; + + beforeEach(async () => { + previousDataDir = process.env.NOTIFY_DATA_DIR; + dir = await mkdtemp(path.join(tmpdir(), "alert-search-cursors-")); + process.env.NOTIFY_DATA_DIR = dir; + closeDatabase(); + }); + + afterEach(async () => { + closeDatabase(); + if (previousDataDir === undefined) { + delete process.env.NOTIFY_DATA_DIR; + } else { + process.env.NOTIFY_DATA_DIR = previousDataDir; + } + await rm(dir, { recursive: true, force: true }); + }); + + it("no existing cursor → retrieval receives no afterId or afterDate", async () => { + const loaded = await loadAlertSearchCursors(USER); + assert.equal(loaded.endorserAfterId, undefined); + assert.equal(loaded.partnerAfterDate, undefined); + + const urls: string[] = []; + const result = await retrieveAlertSearch({ + jwt: JWT, + ...loaded, + endorserBaseUrl: "https://api.endorser.ch", + partnerBaseUrl: "https://partner-api.endorser.ch", + config: { + fetch: async (url) => { + urls.push(url); + if (url.includes("/api/partner/")) { + return jsonResponse(emptyPartnerBody()); + } + return jsonResponse(emptyEndorserBody()); + }, + }, + }); + assert.equal(result.endorser.outcome, "empty"); + for (const url of urls) { + assert.equal(url.includes("afterId="), false); + assert.equal(url.includes("afterDate="), false); + } + }); + + it("complete Endorser retrieval → cursor becomes maximum returned ULID", async () => { + const data = emptyEndorser(); + data.claims = [ + { id: ulid(10), issuedAt: "t", issuer: "did:ethr:0x1" }, + { id: ulid(30), issuedAt: "t", issuer: "did:ethr:0x1" }, + ]; + data.plansNearby = [{ handleId: "p", jwtId: ulid(20) }]; + assert.equal(maxEndorserAlertSearchUlid(data), ulid(30)); + + const advanced = await advanceAlertSearchCursors( + USER, + combined(endorserResult("success", data), partnerResult("empty")) + ); + assert.equal(advanced.endorserAdvanced, true); + assert.equal(advanced.endorserAfterId, ulid(30)); + assert.equal(advanced.partnerAdvanced, false); + }); + + it("Endorser empty retrieval → existing cursor unchanged", async () => { + await alertSearchCursorsDb.setEndorserAfterId(USER, ulid(99)); + const advanced = await advanceAlertSearchCursors( + USER, + combined(endorserResult("empty"), partnerResult("empty")) + ); + assert.equal(advanced.endorserAdvanced, false); + assert.equal(advanced.endorserAfterId, ulid(99)); + }); + + it("Endorser pagination → cursor unchanged", async () => { + await alertSearchCursorsDb.setEndorserAfterId(USER, ulid(5)); + const data = emptyEndorser(); + data.claims = Array.from({ length: ALERT_SEARCH_PAGE_SIZE }, (_, i) => ({ + id: ulid(100 + i), + issuedAt: "t", + issuer: "did:ethr:0x1", + })); + const advanced = await advanceAlertSearchCursors( + USER, + combined(endorserResult("pagination", data), partnerResult("empty")) + ); + assert.equal(advanced.endorserAdvanced, false); + assert.equal(advanced.endorserAfterId, ulid(5)); + }); + + it("Endorser error → cursor unchanged", async () => { + await alertSearchCursorsDb.setEndorserAfterId(USER, ulid(5)); + for (const outcome of ["auth", "timeout", "network", "malformed", "http"] as const) { + const advanced = await advanceAlertSearchCursors( + USER, + combined(endorserResult(outcome), partnerResult("empty")) + ); + assert.equal(advanced.endorserAdvanced, false, outcome); + assert.equal(advanced.endorserAfterId, ulid(5), outcome); + } + }); + + it("complete Partner retrieval → cursor becomes maximum returned updatedAt", async () => { + const data = emptyPartner(); + data.profilesNearby = [ + { + issuerDid: "did:ethr:0x1", + description: "a", + updatedAt: "2026-01-01T00:00:00.000Z", + }, + { + issuerDid: "did:ethr:0x2", + description: "b", + updatedAt: "2026-03-01T12:00:00.000Z", + }, + ]; + const advanced = await advanceAlertSearchCursors( + USER, + combined(endorserResult("empty"), partnerResult("success", data)) + ); + assert.equal(advanced.partnerAdvanced, true); + assert.equal(advanced.partnerAfterAt, "2026-03-01T12:00:00.000Z"); + }); + + it("Partner empty retrieval → existing cursor unchanged", async () => { + await alertSearchCursorsDb.setPartnerAfterAt(USER, "2026-02-01T00:00:00.000Z"); + const advanced = await advanceAlertSearchCursors( + USER, + combined(endorserResult("empty"), partnerResult("empty")) + ); + assert.equal(advanced.partnerAdvanced, false); + assert.equal(advanced.partnerAfterAt, "2026-02-01T00:00:00.000Z"); + }); + + it("Partner pagination → cursor unchanged", async () => { + await alertSearchCursorsDb.setPartnerAfterAt(USER, "2026-02-01T00:00:00.000Z"); + const data = emptyPartner(); + data.profilesNearby = Array.from({ length: ALERT_SEARCH_PAGE_SIZE }, (_, i) => ({ + issuerDid: `did:ethr:0x${i}`, + description: "p", + updatedAt: "2026-04-01T00:00:00.000Z", + })); + const advanced = await advanceAlertSearchCursors( + USER, + combined(endorserResult("empty"), partnerResult("pagination", data)) + ); + assert.equal(advanced.partnerAdvanced, false); + assert.equal(advanced.partnerAfterAt, "2026-02-01T00:00:00.000Z"); + }); + + it("Partner error → cursor unchanged", async () => { + await alertSearchCursorsDb.setPartnerAfterAt(USER, "2026-02-01T00:00:00.000Z"); + const advanced = await advanceAlertSearchCursors( + USER, + combined(endorserResult("empty"), partnerResult("network")) + ); + assert.equal(advanced.partnerAdvanced, false); + assert.equal(advanced.partnerAfterAt, "2026-02-01T00:00:00.000Z"); + }); + + it("Endorser success + Partner failure → only Endorser advances", async () => { + const endorserData = emptyEndorser(); + endorserData.claims = [ + { id: ulid(40), issuedAt: "t", issuer: "did:ethr:0x1" }, + ]; + const advanced = await advanceAlertSearchCursors( + USER, + combined(endorserResult("success", endorserData), partnerResult("auth")) + ); + assert.equal(advanced.endorserAdvanced, true); + assert.equal(advanced.endorserAfterId, ulid(40)); + assert.equal(advanced.partnerAdvanced, false); + assert.equal(advanced.partnerAfterAt, null); + }); + + it("Endorser failure + Partner success → only Partner advances", async () => { + const partnerData = emptyPartner(); + partnerData.profilesNearby = [ + { + issuerDid: "did:ethr:0x1", + description: "p", + updatedAt: "2026-05-01T00:00:00.000Z", + }, + ]; + const advanced = await advanceAlertSearchCursors( + USER, + combined(endorserResult("timeout"), partnerResult("success", partnerData)) + ); + assert.equal(advanced.endorserAdvanced, false); + assert.equal(advanced.endorserAfterId, null); + assert.equal(advanced.partnerAdvanced, true); + assert.equal(advanced.partnerAfterAt, "2026-05-01T00:00:00.000Z"); + }); + + it("both successful → both advance", async () => { + const endorserData = emptyEndorser(); + endorserData.trackedPlanClaims = [ + { id: ulid(7), issuedAt: "t", issuer: "did:ethr:0x1" }, + ]; + const partnerData = emptyPartner(); + partnerData.profilesNearby = [ + { + issuerDid: "did:ethr:0x1", + description: "p", + updatedAt: "2026-06-01T00:00:00.000Z", + }, + ]; + const advanced = await advanceAlertSearchCursors( + USER, + combined( + endorserResult("success", endorserData), + partnerResult("success", partnerData) + ) + ); + assert.equal(advanced.endorserAdvanced, true); + assert.equal(advanced.partnerAdvanced, true); + assert.equal(advanced.endorserAfterId, ulid(7)); + assert.equal(advanced.partnerAfterAt, "2026-06-01T00:00:00.000Z"); + }); + + it("no usable returned cursor → existing cursor unchanged", async () => { + await alertSearchCursorsDb.setEndorserAfterId(USER, ulid(8)); + await alertSearchCursorsDb.setPartnerAfterAt(USER, "2026-01-01T00:00:00.000Z"); + const endorserData = emptyEndorser(); + endorserData.claims = [ + { id: "not-a-ulid", issuedAt: "t", issuer: "did:ethr:0x1" }, + ]; + const partnerData = emptyPartner(); + partnerData.profilesNearby = [ + { issuerDid: "did:ethr:0x1", description: "p" }, + ]; + const advanced = await advanceAlertSearchCursors( + USER, + combined( + endorserResult("success", endorserData), + partnerResult("success", partnerData) + ) + ); + assert.equal(advanced.endorserAdvanced, false); + assert.equal(advanced.partnerAdvanced, false); + assert.equal(advanced.endorserAfterId, ulid(8)); + assert.equal(advanced.partnerAfterAt, "2026-01-01T00:00:00.000Z"); + }); + + it("survives close and reopen of the SQLite connection", async () => { + await alertSearchCursorsDb.setEndorserAfterId(USER, ulid(11)); + await alertSearchCursorsDb.setPartnerAfterAt( + USER, + "2026-07-01T00:00:00.000Z" + ); + closeDatabase(); + + const loaded = await loadAlertSearchCursors(USER); + assert.equal(loaded.endorserAfterId, ulid(11)); + assert.equal(loaded.partnerAfterDate, "2026-07-01T00:00:00.000Z"); + + const other = await loadAlertSearchCursors(OTHER_USER); + assert.equal(other.endorserAfterId, undefined); + assert.equal(other.partnerAfterDate, undefined); + }); + + it("loaded Partner cursor is sent as afterDate, not afterId 0", async () => { + await alertSearchCursorsDb.setPartnerAfterAt( + USER, + "2026-08-01T00:00:00.000Z" + ); + await alertSearchCursorsDb.setEndorserAfterId(USER, ulid(12)); + const loaded = await loadAlertSearchCursors(USER); + const urls: string[] = []; + await retrieveAlertSearch({ + jwt: JWT, + ...loaded, + endorserBaseUrl: "https://api.endorser.ch", + partnerBaseUrl: "https://partner-api.endorser.ch", + config: { + fetch: async (url) => { + urls.push(url); + if (url.includes("/api/partner/")) { + return jsonResponse(emptyPartnerBody()); + } + return jsonResponse(emptyEndorserBody()); + }, + }, + }); + const endorserUrl = urls.find((u) => u.includes("/api/v2/report/")) ?? ""; + const partnerUrl = urls.find((u) => u.includes("/api/partner/")) ?? ""; + assert.equal(endorserUrl.includes(ulid(12)), true); + assert.equal(partnerUrl.includes("afterId=0"), false); + assert.equal( + decodeURIComponent(partnerUrl).includes("afterDate=2026-08-01T00:00:00.000Z"), + true + ); + }); +}); diff --git a/src/alertSearch/cursors.ts b/src/alertSearch/cursors.ts new file mode 100644 index 0000000..0e55338 --- /dev/null +++ b/src/alertSearch/cursors.ts @@ -0,0 +1,137 @@ +import { alertSearchCursorsDb } from "../db/alertSearchCursorsSqlite.js"; +import { isAlertSearchCursorUlid } from "./params.js"; +import type { + AlertSearchSourceResult, + RetrieveAlertSearchResult, +} from "./retrieve.js"; +import type { + EndorserAlertSearchData, + PartnerAlertSearchData, +} from "./types.js"; + +export type StoredAlertSearchCursors = { + endorserAfterId?: string; + partnerAfterDate?: string; +}; + +export type CursorAdvanceResult = { + endorserAdvanced: boolean; + partnerAdvanced: boolean; + endorserAfterId: string | null; + partnerAfterAt: string | null; +}; + +const COMPLETE_OUTCOMES = new Set(["success"]); + +function isCompleteOutcome(outcome: string): boolean { + return COMPLETE_OUTCOMES.has(outcome); +} + +/** MAX ULID across Endorser alertSearch identifier fields. */ +export function maxEndorserAlertSearchUlid( + data: EndorserAlertSearchData +): string | undefined { + const ids: string[] = []; + for (const row of data.claims) { + if (isAlertSearchCursorUlid(row.id)) ids.push(row.id); + } + for (const row of data.personalPlanContributions) { + if (isAlertSearchCursorUlid(row.id)) ids.push(row.id); + } + for (const row of data.trackedPlanClaims) { + if (isAlertSearchCursorUlid(row.id)) ids.push(row.id); + } + for (const row of data.trackedPlanUpdates) { + if (isAlertSearchCursorUlid(row.jwtId)) ids.push(row.jwtId); + } + for (const row of data.plansNearby) { + if (isAlertSearchCursorUlid(row.jwtId)) ids.push(row.jwtId); + } + if (ids.length === 0) return undefined; + ids.sort(); + return ids[ids.length - 1]; +} + +/** MAX profilesNearby.updatedAt (ISO). Not rowId. */ +export function maxPartnerUpdatedAt( + data: PartnerAlertSearchData +): string | undefined { + let best: string | undefined; + let bestMs = Number.NEGATIVE_INFINITY; + for (const row of data.profilesNearby) { + if (typeof row.updatedAt !== "string" || row.updatedAt.length === 0) { + continue; + } + const ms = Date.parse(row.updatedAt); + if (Number.isNaN(ms)) continue; + if (ms > bestMs) { + bestMs = ms; + best = row.updatedAt; + } + } + return best; +} + +export function nextEndorserCursorFromResult( + endorser: AlertSearchSourceResult +): string | undefined { + if (!isCompleteOutcome(endorser.outcome)) return undefined; + return maxEndorserAlertSearchUlid(endorser.data); +} + +export function nextPartnerCursorFromResult( + partner: AlertSearchSourceResult +): string | undefined { + if (!isCompleteOutcome(partner.outcome)) return undefined; + return maxPartnerUpdatedAt(partner.data); +} + +/** + * Load stored cursors for a retrieval query. + * Omits missing values so first run sends no afterId / afterDate / "0". + */ +export async function loadAlertSearchCursors( + userId: string +): Promise { + const row = await alertSearchCursorsDb.get(userId); + const out: StoredAlertSearchCursors = {}; + if (row?.endorserAfterId && isAlertSearchCursorUlid(row.endorserAfterId)) { + out.endorserAfterId = row.endorserAfterId; + } + if (row?.partnerAfterAt && row.partnerAfterAt.length > 0) { + out.partnerAfterDate = row.partnerAfterAt; + } + return out; +} + +/** + * Advance each cursor independently after retrieveAlertSearch. + * Does not persist on empty, pagination, or error outcomes. + */ +export async function advanceAlertSearchCursors( + userId: string, + result: RetrieveAlertSearchResult +): Promise { + let endorserAdvanced = false; + let partnerAdvanced = false; + + const nextEndorser = nextEndorserCursorFromResult(result.endorser); + if (nextEndorser !== undefined) { + await alertSearchCursorsDb.setEndorserAfterId(userId, nextEndorser); + endorserAdvanced = true; + } + + const nextPartner = nextPartnerCursorFromResult(result.partner); + if (nextPartner !== undefined) { + await alertSearchCursorsDb.setPartnerAfterAt(userId, nextPartner); + partnerAdvanced = true; + } + + const stored = await alertSearchCursorsDb.get(userId); + return { + endorserAdvanced, + partnerAdvanced, + endorserAfterId: stored?.endorserAfterId ?? null, + partnerAfterAt: stored?.partnerAfterAt ?? null, + }; +} diff --git a/src/alertSearch/index.ts b/src/alertSearch/index.ts index b29f31d..f573778 100644 --- a/src/alertSearch/index.ts +++ b/src/alertSearch/index.ts @@ -29,3 +29,10 @@ export { } from "./client.js"; export { retrieveAlertSearch } from "./retrieve.js"; export type { RetrieveAlertSearchInput, RetrieveAlertSearchResult } from "./retrieve.js"; +export { + advanceAlertSearchCursors, + loadAlertSearchCursors, + maxEndorserAlertSearchUlid, + maxPartnerUpdatedAt, +} from "./cursors.js"; +export type { CursorAdvanceResult, StoredAlertSearchCursors } from "./cursors.js"; diff --git a/src/alertSearch/retrieve.ts b/src/alertSearch/retrieve.ts index e3ec7be..ec82a9e 100644 --- a/src/alertSearch/retrieve.ts +++ b/src/alertSearch/retrieve.ts @@ -47,8 +47,10 @@ export type RetrieveAlertSearchInput = { jwt: string; /** Endorser cursor from a previous successful daily run. Omit on first run. */ endorserAfterId?: string; - /** Partner cursor from a previous successful daily run. Independent of Endorser. */ + /** Partner afterId ULID (decoded to a timestamp by the server). Prefer partnerAfterDate. */ partnerAfterId?: string; + /** Partner exclusive lower bound: updatedAt > this ISO time. Omit on first run. */ + partnerAfterDate?: string; /** User-selected nearby-search bbox (alertSearch minLoc* / maxLoc* shape). */ location?: AlertSearchLocationBBox; planHandleIds?: string[]; @@ -248,6 +250,7 @@ async function retrieveEndorserPages(input: { async function retrievePartnerPages(input: { jwt: string; afterId?: string; + afterDate?: string; location?: AlertSearchLocationBBox; baseUrl: string; maxPages: number; @@ -261,6 +264,7 @@ async function retrievePartnerPages(input: { for (let page = 0; page < input.maxPages; page++) { const query: AlertSearchQueryInput = { afterId: input.afterId, + afterDate: input.afterDate, beforeDate, location: input.location, }; @@ -346,6 +350,7 @@ export async function retrieveAlertSearch( retrievePartnerPages({ jwt: input.jwt, afterId: input.partnerAfterId, + afterDate: input.partnerAfterDate, location: input.location, baseUrl: partnerBaseUrl, maxPages, diff --git a/src/db/alertSearchCursorsSqlite.ts b/src/db/alertSearchCursorsSqlite.ts new file mode 100644 index 0000000..f9bd26e --- /dev/null +++ b/src/db/alertSearchCursorsSqlite.ts @@ -0,0 +1,93 @@ +import { getDatabase } from "./sqlite.js"; + +export type AlertSearchCursorRecord = { + userId: string; + endorserAfterId: string | null; + partnerAfterAt: string | null; + createdAt: string; + updatedAt: string; +}; + +type CursorDbRow = { + user_id: string; + endorser_after_id: string | null; + partner_after_at: string | null; + created_at: string; + updated_at: string; +}; + +function toRecord(row: CursorDbRow): AlertSearchCursorRecord { + return { + userId: row.user_id, + endorserAfterId: row.endorser_after_id, + partnerAfterAt: row.partner_after_at, + createdAt: row.created_at, + updatedAt: row.updated_at, + }; +} + +function ensureRow(userId: string, now: string): void { + getDatabase() + .prepare( + ` + INSERT INTO alert_search_cursors ( + user_id, endorser_after_id, partner_after_at, created_at, updated_at + ) VALUES (?, NULL, NULL, ?, ?) + ON CONFLICT(user_id) DO NOTHING + ` + ) + .run(userId, now, now); +} + +export const alertSearchCursorsDb = { + async get( + userId: string + ): Promise { + const row = getDatabase() + .prepare( + ` + SELECT user_id, endorser_after_id, partner_after_at, created_at, updated_at + FROM alert_search_cursors + WHERE user_id = ? + ` + ) + .get(userId) as CursorDbRow | undefined; + return row === undefined ? undefined : toRecord(row); + }, + + async setEndorserAfterId(userId: string, afterId: string): Promise { + const now = new Date().toISOString(); + const connection = getDatabase(); + const run = connection.transaction(() => { + ensureRow(userId, now); + connection + .prepare( + ` + UPDATE alert_search_cursors + SET endorser_after_id = ?, updated_at = ? + WHERE user_id = ? + ` + ) + .run(afterId, now, userId); + }); + run(); + }, + + async setPartnerAfterAt(userId: string, afterAt: string): Promise { + const now = new Date().toISOString(); + const connection = getDatabase(); + const run = connection.transaction(() => { + ensureRow(userId, now); + connection + .prepare( + ` + UPDATE alert_search_cursors + SET partner_after_at = ?, updated_at = ? + WHERE user_id = ? + ` + ) + .run(afterAt, now, userId); + }); + run(); + }, +}; diff --git a/src/db/sqlite.ts b/src/db/sqlite.ts index 951cf00..71879d5 100644 --- a/src/db/sqlite.ts +++ b/src/db/sqlite.ts @@ -2,9 +2,9 @@ import { mkdirSync } from "node:fs"; import path from "node:path"; import Database from "better-sqlite3"; -const dataDir = - process.env.NOTIFY_DATA_DIR ?? path.join(process.cwd(), "data"); -const dbFile = path.join(dataDir, "notify.sqlite"); +function dataDirPath(): string { + return process.env.NOTIFY_DATA_DIR ?? path.join(process.cwd(), "data"); +} const SCHEMA_SQL = ` CREATE TABLE IF NOT EXISTS fcm_registrations ( @@ -77,6 +77,14 @@ CREATE INDEX IF NOT EXISTS idx_alert_auth_jwts_user_day CREATE INDEX IF NOT EXISTS idx_alert_auth_jwts_batch_pk ON alert_authorization_jwts (batch_pk); + +CREATE TABLE IF NOT EXISTS alert_search_cursors ( + user_id TEXT PRIMARY KEY NOT NULL, + endorser_after_id TEXT, + partner_after_at TEXT, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL +); `; let database: Database.Database | null = null; @@ -91,8 +99,9 @@ function ensureSchema(connection: Database.Database): void { */ export function getDatabase(): Database.Database { if (database === null) { + const dataDir = dataDirPath(); mkdirSync(dataDir, { recursive: true }); - database = new Database(dbFile); + database = new Database(path.join(dataDir, "notify.sqlite")); database.pragma("journal_mode = WAL"); ensureSchema(database); }