fix(#711): enforce sender ownership on cancel endpoint and fix pre-existing blockers

backend/src/auth.test.ts

@@ -183,7 +183,9 @@ describe('Stream Ownership Enforcement', () => {
 
   beforeAll(async () => {
     const { initDb, getDb } = await import('./services/db');
+    const { initCache } = await import('./services/cache');
     initDb();
+    initCache();
     const db = getDb();
 
     // Clean up any leftover state

backend/src/services/db.ts

@@ -308,6 +308,29 @@ class PostgresDatabase {
   }
 }
 
+export function syncFtsIndex(id: string, sender: string, recipient: string, assetCode: string): void {
+  if (isPostgres()) return;
+  // FTS index sync is currently a no-op; the streams_fts virtual table
+  // is only created for SQLite and is handled separately if needed.
+}
+
+/**
+ * Adds a column to a table if it doesn't already exist.
+ * Safe for both SQLite and Postgres.
+ */
+function addColumnIfMissing(database: any, table: string, column: string, typeDef: string): void {
+  try {
+    const cols = database
+      .prepare(`PRAGMA table_info(${table})`)
+      .all() as Array<{ name: string }>;
+    if (cols.some((c) => c.name === column)) return;
+  } catch {
+    // If PRAGMA fails (e.g. table does not exist), skip.
+    return;
+  }
+  database.exec(`ALTER TABLE ${table} ADD COLUMN ${column} ${typeDef}`);
+}
+
 export function initDb(): void {
   if (isPostgres()) {
     db = new PostgresDatabase(process.env.DATABASE_URL!);
@@ -327,4 +350,8 @@ export function initDb(): void {
   }
 
   runMigrations(db);
+
+  // Incremental schema patches for columns added after the baseline.
+  addColumnIfMissing(db, "streams", "cliff_seconds", "INTEGER NOT NULL DEFAULT 0");
+  addColumnIfMissing(db, "stream_archive", "cliff_seconds", "INTEGER NOT NULL DEFAULT 0");
 }

backend/src/services/streamStore.cancel.integration.test.ts

@@ -3,6 +3,7 @@ import request from "supertest";
 import jwt from "jsonwebtoken";
 import { app } from "../index";
 import { initDb, getDb } from "./db";
+import { initCache } from "./cache";
 import { getStreamHistory } from "./eventHistory";
 import { getJwtSecret } from "./auth";
 import path from "path";
@@ -27,6 +28,7 @@ describe("POST /api/streams/:id/cancel Integration Tests", () => {
 
     // Initialize database
     initDb();
+    initCache();
 
     // Create auth tokens for tests
     authToken = jwt.sign({ accountId: mockSender }, getJwtSecret(), { expiresIn: '1h' });

backend/src/services/streamStore.ts

@@ -809,8 +809,6 @@ export async function createStream(input: StreamInput): Promise<StreamRecord> {
     const op = createStreamOperation(contractId, input, startAt);
 
     const txToSimulate = new TransactionBuilder(sourceAccount, {
-  const built = await rpcServer.prepareTransaction(
-    new TransactionBuilder(sourceAccount, {
       fee: "1000",
       networkPassphrase: netPass,
     })