feat(tags): add stream tags/labels for organization (#727)

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;

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;

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;

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,20 +162,22 @@ describe("database migrations", () => {
 
     runMigrations(db);
 
-    rollbackMigration(db, 4);
+    rollbackMigration(db, 5);
 
     expect(getTableColumns(db, "webhook_dead_letters")).toEqual([
       "id",
       "url",
       "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]);
   });
 });

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<string, string> | 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,15 +138,16 @@ function rowToRecord(row: StreamRow): StreamRecord {
     pausedDuration: row.paused_duration ?? 0,
     cliffSeconds: row.cliff_seconds ?? 0,
     metadata,
+    tags,
   };
 }
 
 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<StreamRecord> {
     createdAt: nowInSeconds(),
     pausedDuration: 0,
     cliffSeconds: input.cliffSeconds ?? 0,
+    tags: input.tags ?? null,
   };
 
   const db = getDb();
@@ -983,8 +1001,8 @@ export async function archiveOldStreams(): Promise<number> {
 
         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<number> {
           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);

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) {