diff --git a/backend/migrations/005_add_stream_tags.down.sql b/backend/migrations/005_add_stream_tags.down.sql new file mode 100644 index 00000000..b0b77233 --- /dev/null +++ b/backend/migrations/005_add_stream_tags.down.sql @@ -0,0 +1,2 @@ +ALTER TABLE streams DROP COLUMN tags; +ALTER TABLE stream_archive DROP COLUMN tags; diff --git a/backend/migrations/005_add_stream_tags.sql b/backend/migrations/005_add_stream_tags.sql new file mode 100644 index 00000000..2425a21a --- /dev/null +++ b/backend/migrations/005_add_stream_tags.sql @@ -0,0 +1,2 @@ +ALTER TABLE streams ADD COLUMN tags TEXT; +ALTER TABLE stream_archive ADD COLUMN tags TEXT; diff --git a/backend/src/index.ts b/backend/src/index.ts index 4ff0cfdc..6e7a8076 100644 --- a/backend/src/index.ts +++ b/backend/src/index.ts @@ -145,6 +145,7 @@ const listStreamsQuerySchema = z.object({ return value.split(",").map((code) => code.trim().toUpperCase()); }), q: z.string().trim().optional(), + tag: z.string().trim().optional(), minAmount: z.coerce .number() .nonnegative("minAmount must be a non-negative number") @@ -591,6 +592,12 @@ app.get("/api/streams", readLimiter, async (req: Request, res: Response) => { if (query.maxAmount !== undefined) { data = data.filter((stream) => stream.totalAmount <= query.maxAmount!); } + if (query.tag) { + const filterTag = query.tag.toLowerCase(); + data = data.filter((stream) => + stream.tags && stream.tags.some((t) => t.toLowerCase() === filterTag), + ); + } const total = data.length; const page = query.page ?? PAGINATION_DEFAULT_PAGE; @@ -897,6 +904,12 @@ app.get( if (query.maxAmount !== undefined) { data = data.filter((stream) => stream.totalAmount <= query.maxAmount!); } + if (query.tag) { + const filterTag = query.tag.toLowerCase(); + data = data.filter((stream) => + stream.tags && stream.tags.some((t) => t.toLowerCase() === filterTag), + ); + } const hasPage = req.query.page !== undefined; const hasLimit = req.query.limit !== undefined; @@ -981,6 +994,12 @@ app.get( if (query.maxAmount !== undefined) { data = data.filter((stream) => stream.totalAmount <= query.maxAmount!); } + if (query.tag) { + const filterTag = query.tag.toLowerCase(); + data = data.filter((stream) => + stream.tags && stream.tags.some((t) => t.toLowerCase() === filterTag), + ); + } const hasPage = req.query.page !== undefined; const hasLimit = req.query.limit !== undefined; @@ -1081,6 +1100,12 @@ app.get( if (query.maxAmount !== undefined) { data = data.filter((stream) => stream.totalAmount <= query.maxAmount!); } + if (query.tag) { + const filterTag = query.tag.toLowerCase(); + data = data.filter((stream) => + stream.tags && stream.tags.some((t) => t.toLowerCase() === filterTag), + ); + } const hasPage = req.query.page !== undefined; const hasLimit = req.query.limit !== undefined; @@ -1156,6 +1181,12 @@ app.get( ); }); } + if (query.tag) { + const filterTag = query.tag.toLowerCase(); + data = data.filter((stream) => + stream.tags && stream.tags.some((t) => t.toLowerCase() === filterTag), + ); + } const hasPage = req.query.page !== undefined; const hasLimit = req.query.limit !== undefined; diff --git a/backend/src/services/migrations.test.ts b/backend/src/services/migrations.test.ts index 900f469f..b6d537e7 100644 --- a/backend/src/services/migrations.test.ts +++ b/backend/src/services/migrations.test.ts @@ -27,6 +27,7 @@ const EXPECTED_STREAMS_COLUMNS = [ "paused_at", "paused_duration", "metadata", + "tags", ]; const EXPECTED_WEBHOOK_DEAD_LETTERS_COLUMNS = [ @@ -161,7 +162,7 @@ describe("database migrations", () => { runMigrations(db); - rollbackMigration(db, 4); + rollbackMigration(db, 5); expect(getTableColumns(db, "webhook_dead_letters")).toEqual([ "id", @@ -169,12 +170,14 @@ describe("database migrations", () => { "payload", "last_error", "failed_at", + "stream_id", + "event", ]); const applied = db .prepare("SELECT version FROM schema_migrations ORDER BY version") .all() as Array<{ version: number }>; - expect(applied.map((row) => row.version)).toEqual([1, 2, 3]); + expect(applied.map((row) => row.version)).toEqual([1, 2, 3, 4]); }); }); diff --git a/backend/src/services/streamStore.ts b/backend/src/services/streamStore.ts index 71eb6d05..353a9608 100644 --- a/backend/src/services/streamStore.ts +++ b/backend/src/services/streamStore.ts @@ -34,6 +34,7 @@ export interface StreamInput { durationSeconds: number; startAt?: number; cliffSeconds?: number; + tags?: string[]; } export interface StreamFeeEstimate { @@ -57,6 +58,7 @@ export interface StreamRecord { pausedDuration: number; cliffSeconds: number; metadata?: Record | null; + tags?: string[] | null; } export interface StreamProgress { @@ -97,6 +99,7 @@ interface StreamRow { paused_duration: number; cliff_seconds: number; metadata: string | null; + tags: string | null; } function rowToRecord(row: StreamRow): StreamRecord { @@ -108,6 +111,17 @@ function rowToRecord(row: StreamRow): StreamRecord { metadata = null; } } + let tags: string[] | null = null; + if (row.tags) { + try { + const parsed = JSON.parse(row.tags); + if (Array.isArray(parsed)) { + tags = parsed; + } + } catch { + tags = null; + } + } return { id: row.id, sender: row.sender, @@ -124,6 +138,7 @@ function rowToRecord(row: StreamRow): StreamRecord { pausedDuration: row.paused_duration ?? 0, cliffSeconds: row.cliff_seconds ?? 0, metadata, + tags, }; } @@ -131,8 +146,8 @@ function upsertStream(record: StreamRecord): void { const db = getDb(); db.prepare( ` - INSERT INTO streams (id, sender, recipient, asset_code, total_amount, duration_seconds, start_at, created_at, canceled_at, completed_at, refunded_amount, archived_at, paused_at, paused_duration, cliff_seconds, metadata) - VALUES (@id, @sender, @recipient, @assetCode, @totalAmount, @durationSeconds, @startAt, @createdAt, @canceledAt, @completedAt, @refundedAmount, @archivedAt, @pausedAt, @pausedDuration, @cliffSeconds, @metadata) + INSERT INTO streams (id, sender, recipient, asset_code, total_amount, duration_seconds, start_at, created_at, canceled_at, completed_at, refunded_amount, archived_at, paused_at, paused_duration, cliff_seconds, metadata, tags) + VALUES (@id, @sender, @recipient, @assetCode, @totalAmount, @durationSeconds, @startAt, @createdAt, @canceledAt, @completedAt, @refundedAmount, @archivedAt, @pausedAt, @pausedDuration, @cliffSeconds, @metadata, @tags) ON CONFLICT(id) DO UPDATE SET sender = excluded.sender, recipient = excluded.recipient, @@ -148,7 +163,8 @@ function upsertStream(record: StreamRecord): void { paused_at = excluded.paused_at, paused_duration = excluded.paused_duration, cliff_seconds = excluded.cliff_seconds, - metadata = excluded.metadata + metadata = excluded.metadata, + tags = excluded.tags `, ).run({ id: record.id, @@ -167,6 +183,7 @@ function upsertStream(record: StreamRecord): void { pausedDuration: record.pausedDuration ?? 0, cliffSeconds: record.cliffSeconds ?? 0, metadata: record.metadata ? JSON.stringify(record.metadata) : null, + tags: record.tags ? JSON.stringify(record.tags) : null, }); syncFtsIndex(record.id, record.sender, record.recipient, record.assetCode); } @@ -859,6 +876,7 @@ export async function createStream(input: StreamInput): Promise { createdAt: nowInSeconds(), pausedDuration: 0, cliffSeconds: input.cliffSeconds ?? 0, + tags: input.tags ?? null, }; const db = getDb(); @@ -983,8 +1001,8 @@ export async function archiveOldStreams(): Promise { db.prepare( ` - INSERT INTO stream_archive (id, sender, recipient, asset_code, total_amount, duration_seconds, start_at, created_at, canceled_at, completed_at, refunded_amount, archived_at) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + INSERT INTO stream_archive (id, sender, recipient, asset_code, total_amount, duration_seconds, start_at, created_at, canceled_at, completed_at, refunded_amount, archived_at, tags) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) `, ).run( record.id, @@ -999,6 +1017,7 @@ export async function archiveOldStreams(): Promise { record.completedAt ?? null, record.refundedAmount ?? null, now, + record.tags ? JSON.stringify(record.tags) : null, ); db.prepare("UPDATE streams SET archived_at = ? WHERE id = ?").run(now, record.id); diff --git a/backend/src/validation/schemas.ts b/backend/src/validation/schemas.ts index cd03cffb..02ff78f3 100644 --- a/backend/src/validation/schemas.ts +++ b/backend/src/validation/schemas.ts @@ -53,6 +53,11 @@ export const unixTimestampSchema = z.coerce .int("startAt must be a valid UNIX timestamp in seconds.") .positive("startAt must be a valid UNIX timestamp in seconds."); +export const tagsSchema = z + .array(z.string().trim().min(1, "Tag must not be empty").max(50, "Tag must be 50 characters or fewer")) + .max(5, "A stream can have at most 5 tags") + .optional(); + export const createStreamPayloadSchema = z .object({ sender: stellarAccountIdSchema, @@ -62,6 +67,7 @@ export const createStreamPayloadSchema = z durationSeconds: durationSecondsSchema, startAt: unixTimestampSchema.optional(), cliffSeconds: z.coerce.number().int().nonnegative().optional(), + tags: tagsSchema, }) .superRefine((payload, ctx) => { if (payload.sender === payload.recipient) {