fix: stream cancel visibility, wizard tx hash & post-submit stream matching

backend/src/controllers/stream/cancel.ts

@@ -89,8 +89,10 @@ export const cancelStreamHandler = async (req: AuthenticatedRequest, res: Respon
 
     const txHash = await sorobanService.cancelStream(parsedStreamId, senderSecret);
 
-    // 5. Update DB record status using repository helper
-    await streamRepository.updateStatus(parsedStreamId, 'CANCELLED');
+    // 5. Update DB record status and write a provisional CANCELLED StreamEvent
+    // atomically (issue #1219) so the stream shows up under the "cancelled"
+    // filter before the indexer has processed the on-chain event.
+    await streamRepository.markCancelledProvisionally(parsedStreamId, txHash);
 
     logger.info(`[CancelStream] Stream ${parsedStreamId} cancelled by ${callerAddress}. Tx: ${txHash}`);
 

backend/src/repositories/stream.repository.ts

@@ -1,4 +1,68 @@
 import { prisma, withReplicaFallback } from '../lib/prisma.js';
+import type { Prisma } from '../generated/prisma/index.js';
+
+/**
+ * Transaction client passed to interactive Prisma transactions.
+ */
+export type PrismaTxClient = Prisma.TransactionClient;
+
+/**
+ * Atomically mark a stream as cancelled (via the API) and write a provisional
+ * CANCELLED StreamEvent in the same transaction.
+ *
+ * The indexer's handleStreamCancelled writes an authoritative event for the
+ * real on-chain StreamCancelled event later; both variants share the
+ * (transactionHash, eventType) unique key, so the indexer reconciles (updates)
+ * the provisional record instead of creating a duplicate.
+ *
+ * Issue #1219: without the provisional event the cancelled stream matched
+ * neither the "active" filter (isActive=false) nor the "cancelled" filter
+ * (requires a CANCELLED event) and disappeared from all status views until
+ * the on-chain event was indexed.
+ */
+export const markCancelledProvisionally = async (
+  streamId: bigint,
+  txHash: string,
+  txClient?: PrismaTxClient,
+) => {
+  const timestamp = Math.floor(Date.now() / 1000);
+  const run = async (tx: PrismaTxClient) => {
+    await tx.stream.update({
+      where: { streamId },
+      data: {
+        isActive: false,
+      },
+    });
+
+    await tx.streamEvent.upsert({
+      where: {
+        transactionHash_eventType: {
+          transactionHash: txHash,
+          eventType: 'CANCELLED',
+        },
+      },
+      create: {
+        streamId,
+        eventType: 'CANCELLED',
+        amount: null,
+        transactionHash: txHash,
+        // Ledger details are unknown until the indexer sees the real event;
+        // ledgerSequence 0 + provisional metadata marks this as API-written.
+        ledgerSequence: 0,
+        timestamp,
+        metadata: JSON.stringify({ provisional: true, source: 'api-cancel' }),
+      },
+      // If another request already wrote the provisional event for this
+      // transaction, don't clobber it with a second provisional copy.
+      update: {},
+    });
+  };
+
+  if (txClient) {
+    return run(txClient);
+  }
+  return prisma.$transaction(run);
+};
 
 /**
  * Update the status and active flag of a stream in the database.

backend/src/workers/soroban-event-worker.ts

@@ -1166,31 +1166,35 @@ export class SorobanEventWorker {
       });
       if (existingEvent) {
         logger.warn(
-          `[SorobanWorker] Duplicate StreamEvent skipped: txHash=${event.txHash} type=CANCELLED`,
+          `[SorobanWorker] Reconciling existing StreamEvent: txHash=${event.txHash} type=CANCELLED`,
         );
-      } else {
-        await tx.streamEvent.upsert({
-          where: {
-            transactionHash_eventType: {
-              transactionHash: event.txHash,
-              eventType: "CANCELLED",
-            },
-          },
-          create: {
-            streamId,
-            eventType: "CANCELLED",
-            amount: refundedAmount,
+      }
+      await tx.streamEvent.upsert({
+        where: {
+          transactionHash_eventType: {
             transactionHash: event.txHash,
-            ledgerSequence: event.ledger,
-            timestamp,
-            metadata: JSON.stringify({ amountWithdrawn, refundedAmount }),
-          },
-          update: {
-            ledgerSequence: event.ledger,
-            timestamp,
+            eventType: "CANCELLED",
           },
-        });
-      }
+        },
+        create: {
+          streamId,
+          eventType: "CANCELLED",
+          amount: refundedAmount,
+          transactionHash: event.txHash,
+          ledgerSequence: event.ledger,
+          timestamp,
+          metadata: JSON.stringify({ amountWithdrawn, refundedAmount }),
+        },
+        // When an API-cancelled stream already wrote a provisional CANCELLED
+        // event (issue #1219), replace the provisional fields with the
+        // authoritative ledger details from the indexed on-chain event.
+        update: {
+          ledgerSequence: event.ledger,
+          timestamp,
+          amount: refundedAmount,
+          metadata: JSON.stringify({ amountWithdrawn, refundedAmount }),
+        },
+      });
     });
 
     sseService.broadcastToStream(String(streamId), "stream.cancelled", {

backend/tests/cancel.controller.test.ts

@@ -20,6 +20,7 @@ vi.mock('../src/services/sorobanService.js', () => ({
 
 vi.mock('../src/repositories/stream.repository.js', () => ({
   updateStatus: vi.fn(),
+  markCancelledProvisionally: vi.fn(),
 }));
 
 vi.mock('../src/logger.js', () => ({
@@ -79,7 +80,7 @@ describe('Cancel Stream Controller', () => {
     await cancelStreamHandler(req as AuthenticatedRequest, res as Response);
 
     expect(sorobanService.cancelStream).toHaveBeenCalledWith(123n, 'SABC123');
-    expect(streamRepository.updateStatus).toHaveBeenCalledWith(123n, 'CANCELLED');
+    expect(streamRepository.markCancelledProvisionally).toHaveBeenCalledWith(123n, 'tx_hash_123');
     expect(res.status).toHaveBeenCalledWith(200);
     expect(res.json).toHaveBeenCalledWith(expect.objectContaining({ status: 'CANCELLED', txHash: 'tx_hash_123' }));
   });
@@ -103,7 +104,7 @@ describe('Cancel Stream Controller', () => {
     await cancelStreamHandler(req as AuthenticatedRequest, res as Response);
 
     expect(sorobanService.cancelStream).toHaveBeenCalledWith(123n, 'SABC123');
-    expect(streamRepository.updateStatus).not.toHaveBeenCalled();
+    expect(streamRepository.markCancelledProvisionally).not.toHaveBeenCalled();
     expect(res.status).toHaveBeenCalledWith(500);
   });
 });