soroban-event-worker.ts

import { randomUUID } from "crypto";
import { rpc, xdr, StrKey } from "@stellar/stellar-sdk";
import { prisma } from "../lib/prisma.js";
import { INDEXER_STATE_ID, ensureIndexerState } from "../lib/indexer-state.js";
import { sseService } from "../services/sse.service.js";
import logger, { requestContext } from "../logger.js";
import { Prisma } from "../generated/prisma/index.js";
import "../lib/stream-id.js";
import { rpcPool } from "../lib/rpc-pool.js";

// ─── Config ──────────────────────────────────────────────────────────────────

// ─── XDR Decoding Helpers ────────────────────────────────────────────────────

/** Decode an ScVal symbol to a string. */
export function decodeSymbol(val: xdr.ScVal): string {
  return val.sym().toString();
}

/**
 * Decode an ScVal U64 to a JavaScript bigint.
 * `xdr.UInt64` extends Long; `.toString()` gives the decimal representation.
 */
export function decodeU64(val: xdr.ScVal): bigint {
  return BigInt(val.u64().toString());
}

/** Decode an ScVal U32 to a JavaScript number. */
export function decodeU32(val: xdr.ScVal): number {
  return val.u32();
}

/**
 * Decode an ScVal I128 to a decimal string suitable for DB storage.
 * I128 in XDR is split into hi (signed Int64) and lo (unsigned Uint64).
 * Full value = hi * 2^64 + lo.
 */
export function decodeI128(val: xdr.ScVal): string {
  const parts = val.i128();
  const hi = BigInt.asIntN(64, BigInt(parts.hi().toString()));
  const lo = BigInt.asUintN(64, BigInt(parts.lo().toString()));
  return ((hi << 64n) | lo).toString();
}

/**
 * Decode an ScVal Address to a Stellar public key (G...) or contract (C...)
 * string.
 */
export function decodeAddress(val: xdr.ScVal): string {
  const addr = val.address();
  if (addr.switch().value === xdr.ScAddressType.scAddressTypeAccount().value) {
    return StrKey.encodeEd25519PublicKey(addr.accountId().ed25519());
  }
  // addr.contractId() returns a Hash (Opaque[]); cast to Uint8Array for encodeContract
  const hash = addr.contractId();
  return StrKey.encodeContract(Buffer.from(hash as unknown as Uint8Array));
}

/**
 * Decode an ScVal Map (a `#[contracttype]` struct) into a plain object keyed
 * by field name with raw ScVal values for further decoding.
 */
export function decodeMap(val: xdr.ScVal): Record<string, xdr.ScVal> {
  const result: Record<string, xdr.ScVal> = {};
  const entries = val.map();
  if (!entries) return result;
  for (const entry of entries) {
    result[entry.key().sym().toString()] = entry.val();
  }
  return result;
}

// ─── Event-processing counters / degraded signal ─────────────────────────────

/** Sliding window used to decide whether recent failures count as a "spike". */
const FAILURE_WINDOW_MS = 5 * 60 * 1000;
/** Need at least this many attempts in the window before marking degraded. */
const MIN_SAMPLES_FOR_DEGRADED = 3;
/** Degraded when recent failure rate is at or above this threshold. */
const FAILURE_RATE_THRESHOLD = 0.5;

export interface IndexerEventCounters {
  eventsProcessed: number;
  eventsFailed: number;
  lastErrorAt: string | null;
  /** True when recent failure rate indicates a spike (not just lifetime totals). */
  degraded: boolean;
}

// ─── Worker Class ─────────────────────────────────────────────────────────────

export class SorobanEventWorker {
  private readonly contractId: string;
  private readonly server: rpc.Server;
  private readonly pollIntervalMs: number;
  private readonly startLedger: number;

  private isRunning = false;
  private pollTimer: NodeJS.Timeout | undefined;
  /**
   * In-flight fetch/process work from either the scheduler or `triggerPoll`.
   * `waitForDrain()` awaits this so graceful shutdown covers replay batches too.
   */
  private activeBatch: Promise<void> | null = null;
  /** Promise chain that serializes all `fetchAndProcessEvents` invocations. */
  private batchMutex: Promise<void> = Promise.resolve();

  /** Lifetime count of events that processed without throwing. */
  private eventsProcessed = 0;
  /** Lifetime count of events that threw during processing. */
  private eventsFailed = 0;
  /** Timestamp of the most recent per-event processing failure. */
  private lastErrorAt: Date | null = null;
  /** Recent attempt outcomes for sliding-window spike detection. */
  private recentOutcomes: { ok: boolean; at: number }[] = [];

  constructor() {
    this.contractId = process.env.STREAM_CONTRACT_ID ?? "";
    const rpcUrl = process.env.SOROBAN_RPC_URL ?? "https://soroban-testnet.stellar.org";
    this.server = new rpc.Server(rpcUrl, { allowHttp: true });
    this.pollIntervalMs = parseInt(
      process.env.INDEXER_POLL_INTERVAL_MS ?? "5000",
      10,
    );
    this.startLedger = parseInt(process.env.INDEXER_START_LEDGER ?? "0", 10);
  }

  /**
   * Snapshot of event-processing counters for /health and admin metrics.
   * `degraded` is true when ≥50% of attempts in the last 5 minutes failed
   * (with at least 3 samples), so a broken indexer fails the health check
   * even when lag stays low because `updatedAt` is bumped every poll.
   */
  getEventCounters(): IndexerEventCounters {
    return {
      eventsProcessed: this.eventsProcessed,
      eventsFailed: this.eventsFailed,
      lastErrorAt: this.lastErrorAt?.toISOString() ?? null,
      degraded: this.isFailureSpike(),
    };
  }

  /** @internal Reset counters — used by unit tests. */
  resetEventCounters(): void {
    this.eventsProcessed = 0;
    this.eventsFailed = 0;
    this.lastErrorAt = null;
    this.recentOutcomes = [];
  }

  private recordOutcome(ok: boolean): void {
    const now = Date.now();
    this.recentOutcomes.push({ ok, at: now });
    const cutoff = now - FAILURE_WINDOW_MS;
    this.recentOutcomes = this.recentOutcomes.filter((o) => o.at >= cutoff);
  }

  private isFailureSpike(): boolean {
    const cutoff = Date.now() - FAILURE_WINDOW_MS;
    const recent = this.recentOutcomes.filter((o) => o.at >= cutoff);
    if (recent.length < MIN_SAMPLES_FOR_DEGRADED) return false;
    const failed = recent.filter((o) => !o.ok).length;
    return failed / recent.length >= FAILURE_RATE_THRESHOLD;
  }

  /**
   * Start the polling worker. If `STREAM_CONTRACT_ID` is not configured the
   * worker logs a warning and exits gracefully instead of throwing.
   */
  async start(): Promise<void> {
    if (!this.contractId) {
      logger.warn(
        "[SorobanWorker] STREAM_CONTRACT_ID is not set — event indexing disabled.",
      );
      return;
    }

    this.isRunning = true;
    logger.info("[SorobanWorker] Starting Soroban event indexer…");
    await this.poll();
  }

  /** Stop the worker gracefully. */
  stop(): void {
    this.isRunning = false;
    if (this.pollTimer) {
      clearTimeout(this.pollTimer);
      this.pollTimer = undefined;
    }
    logger.info("[SorobanWorker] Stopped.");
  }

  /** Wait for the currently-running poll/replay batch to finish (no-op if idle). */
  async waitForDrain(): Promise<void> {
    if (this.activeBatch) await this.activeBatch;
  }

  /**
   * Trigger an immediate poll cycle (used for replay and manual updates).
   * Serialized with the scheduled poll via `runExclusive` so two cursor writes
   * cannot overlap and regress `lastCursor` (#843).
   *
   * @param customRequestId Optional correlation ID to bind logs to a specific request/replay.
   * @returns The correlation requestId associated with this poll batch.
   */
  async triggerPoll(customRequestId?: string): Promise<string> {
    if (!this.isRunning) {
      return customRequestId || requestContext?.getStore?.()?.requestId || randomUUID();
    }

    const requestId =
      customRequestId || requestContext?.getStore?.()?.requestId || randomUUID();

    try {
      await this.runExclusive(() => {
        const runBatch = () => this.fetchAndProcessEvents();
        return requestContext && typeof requestContext.run === 'function'
          ? requestContext.run({ requestId }, runBatch)
          : runBatch();
      });
    } catch (err) {
      logger.error("[SorobanWorker] Manual poll error:", err);
    }

    return requestId;
  }

  // ─── Internal ──────────────────────────────────────────────────────────────

  /**
   * Run `fn` exclusively with any other poll/replay batch.
   * Registers the work on `activeBatch` so `waitForDrain` awaits replays too.
   */
  private runExclusive(fn: () => Promise<void>): Promise<void> {
    const run = this.batchMutex.then(fn);
    // Keep the mutex chain alive even when a batch rejects.
    const gate = run.then(
      () => undefined,
      () => undefined,
    );
    this.batchMutex = gate;
    this.activeBatch = gate;
    // Clear only if nothing newer registered on activeBatch after this gate.
    void gate.then(() => {
      if (this.activeBatch === gate) {
        this.activeBatch = null;
      }
    });
    return run;
  }

  private scheduleNext(): void {
    if (!this.isRunning) return;
    this.pollTimer = setTimeout(() => this.poll(), this.pollIntervalMs);
  }

  private async ensureSystemStream(
    tx: Prisma.TransactionClient,
  ): Promise<void> {
    const systemUser =
      "GAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAWHF";
    await tx.user.upsert({
      where: { publicKey: systemUser },
      create: { publicKey: systemUser },
      update: {},
    });
    await tx.stream.upsert({
      where: { streamId: 0n },
      create: {
        streamId: 0n,
        sender: systemUser,
        recipient: systemUser,
        tokenAddress:
          "CDAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAHF",
        ratePerSecond: "0",
        depositedAmount: "0",
        withdrawnAmount: "0",
        startTime: 0,
        lastUpdateTime: 0,
        endTime: 0,
        isActive: false,
      },
      update: {},
    });
  }

  private async poll(): Promise<void> {
    try {
      const requestId = randomUUID();
      await this.runExclusive(() => {
        const execute = () =>
          this.fetchAndProcessEvents().catch((err) => {
            logger.error("[SorobanWorker] Unhandled error during poll:", err);
          });
        return requestContext && typeof requestContext.run === 'function'
          ? requestContext.run({ requestId }, execute)
          : execute();
      });
    } finally {
      this.scheduleNext();
    }
  }

  /**
   * Fetch a batch of events from the Soroban RPC starting from the last known
   * cursor (or start ledger on first run) and process each one in order.
   */
  private async fetchAndProcessEvents(): Promise<void> {
    const currentCtx = requestContext?.getStore?.();
    if (!currentCtx?.requestId && requestContext && typeof requestContext.run === 'function') {
      const requestId = randomUUID();
      return requestContext.run({ requestId }, () =>
        this.fetchAndProcessEvents(),
      );
    }

    // Ensure an IndexerState row exists on first run.
    const state = await ensureIndexerState(this.startLedger);

    const baseFilter = {
      filters: [
        {
          type: "contract" as const,
          contractIds: [this.contractId],
        },
      ],
      limit: 100,
    } satisfies Omit<
      Parameters<rpc.Server["getEvents"]>[0],
      "startLedger" | "cursor"
    >;

    // Prefer cursor-based pagination after the first poll so we never
    // re-process events.
    const params: Parameters<rpc.Server["getEvents"]>[0] = state.lastCursor
      ? { ...baseFilter, cursor: state.lastCursor }
      : { ...baseFilter, startLedger: state.lastLedger || this.startLedger };

    const response = await (this.server ? this.server.getEvents(params) : rpcPool.execute("getEvents", (server) => server.getEvents(params)));

    if (response.events.length === 0) return;

    let lastCursor: string | null = state.lastCursor;
    let lastLedger: number = state.lastLedger;
    let hasError = false;

    // Sort events so that 'stream_created' events are processed first in the batch.
    // This ensures that subsequent events (like 'fee_collected') that depend on
    // the stream existing in the DB can find it.
    const sortedEvents = [...response.events].sort((a, b) => {
      const aType = a.topic[0] ? decodeSymbol(a.topic[0]) : "";
      const bType = b.topic[0] ? decodeSymbol(b.topic[0]) : "";
      if (aType === "stream_created" && bType !== "stream_created") return -1;
      if (bType === "stream_created" && aType !== "stream_created") return 1;
      return 0;
    });

    for (const event of sortedEvents) {
      // Only process events from successful contract calls.
      if (!event.inSuccessfulContractCall) continue;

      try {
        await this.processEvent(event);
        this.eventsProcessed += 1;
        this.recordOutcome(true);
        if (!hasError) {
          // Use the event ID as the cursor if pagingToken is not available
          lastCursor = event.id;
          lastLedger = event.ledger;
        }
      } catch (err) {
        hasError = true;
        this.eventsFailed += 1;
        this.lastErrorAt = new Date();
        this.recordOutcome(false);
        logger.error(
          `[SorobanWorker] Failed to process event ${event.id}:`,
          err,
        );
        // Continue processing subsequent events rather than halting.
      }
    }

    // Use the response's final cursor if provided and no error occurred, otherwise the last valid event's ID
    const finalCursor = hasError
      ? lastCursor
      : ((response as any).latestCursor || lastCursor);

    await prisma.indexerState.upsert({
      where: { id: INDEXER_STATE_ID },
      create: {
        id: INDEXER_STATE_ID,
        lastLedger,
        lastCursor: finalCursor,
      },
      update: { lastLedger, lastCursor: finalCursor },
    });

    logger.info(
      `[SorobanWorker] Processed ${response.events.length} event(s) — latest ledger: ${lastLedger}`,
    );
  }

  /**
   * Dispatch a single contract event to the appropriate handler based on the
   * first topic symbol.
   */
  public async processEvent(event: rpc.Api.EventResponse): Promise<void> {
    if (!event.topic || event.topic.length < 1) return;

    const topic0: xdr.ScVal | undefined = event.topic[0];
    if (!topic0) return;

    const eventName = decodeSymbol(topic0);

    if (
      eventName === "fee_config_updated" ||
      eventName === "admin_transferred"
    ) {
      if (eventName === "fee_config_updated") {
        await this.handleFeeConfigUpdated(event);
      } else {
        await this.handleAdminTransferred(event);
      }
      return;
    }

    if (event.topic.length < 2) return;
    const topic1: xdr.ScVal | undefined = event.topic[1];
    if (!topic1) return;

    switch (eventName) {
      case "stream_created":
        await this.handleStreamCreated(event, topic1);
        break;
      case "stream_topped_up":
        await this.handleStreamToppedUp(event, topic1);
        break;
      case "tokens_withdrawn":
        await this.handleTokensWithdrawn(event, topic1);
        break;
      case "stream_paused":
        await this.handleStreamPaused(event, topic1);
        break;
      case "stream_resumed":
        await this.handleStreamResumed(event, topic1);
        break;
      case "stream_cancelled":
        await this.handleStreamCancelled(event, topic1);
        break;
      case "stream_completed":
        await this.handleStreamCompleted(event, topic1);
        break;
      case "fee_collected":
        await this.handleFeeCollected(event, topic1);
        break;
      default:
        // Unrecognised event — ignore silently.
        break;
    }
  }

  private async handleFeeConfigUpdated(
    event: rpc.Api.EventResponse,
  ): Promise<void> {
    const body = decodeMap(event.value);

    if (
      !body["admin"] ||
      !body["old_treasury"] ||
      !body["new_treasury"] ||
      body["old_fee_rate_bps"] === undefined ||
      body["new_fee_rate_bps"] === undefined
    ) {
      throw new Error("FeeConfigUpdated: missing body fields");
    }

    const admin = decodeAddress(body["admin"]);
    const oldTreasury = decodeAddress(body["old_treasury"]);
    const newTreasury = decodeAddress(body["new_treasury"]);
    const oldFeeRateBps = decodeU32(body["old_fee_rate_bps"]);
    const newFeeRateBps = decodeU32(body["new_fee_rate_bps"]);
    const timestamp = Math.floor(Date.now() / 1000);

    await prisma.$transaction(async (tx: Prisma.TransactionClient) => {
      await this.ensureSystemStream(tx);

      await tx.streamEvent.upsert({
        where: {
          transactionHash_eventType: {
            transactionHash: event.txHash,
            eventType: "FEE_CONFIG_UPDATED",
          },
        },
        create: {
          streamId: 0n,
          eventType: "FEE_CONFIG_UPDATED",
          transactionHash: event.txHash,
          ledgerSequence: event.ledger,
          timestamp,
          metadata: JSON.stringify({
            admin,
            old_treasury: oldTreasury,
            new_treasury: newTreasury,
            old_fee_rate_bps: oldFeeRateBps,
            new_fee_rate_bps: newFeeRateBps,
          }),
        },
        update: {
          ledgerSequence: event.ledger,
          timestamp,
        },
      });
    });

    sseService.broadcastToAdmin("stream.fee_config_updated", {
      admin,
      oldTreasury,
      newTreasury,
      oldFeeRateBps,
      newFeeRateBps,
      transactionHash: event.txHash,
      ledger: event.ledger,
      timestamp,
    });
  }

  private async handleAdminTransferred(
    event: rpc.Api.EventResponse,
  ): Promise<void> {
    const body = decodeMap(event.value);

    if (!body["previous_admin"] || !body["new_admin"]) {
      throw new Error("AdminTransferred: missing body fields");
    }

    const previousAdmin = decodeAddress(body["previous_admin"]);
    const newAdmin = decodeAddress(body["new_admin"]);
    const timestamp = Math.floor(Date.now() / 1000);

    await prisma.$transaction(async (tx: Prisma.TransactionClient) => {
      await this.ensureSystemStream(tx);

      await tx.streamEvent.upsert({
        where: {
          transactionHash_eventType: {
            transactionHash: event.txHash,
            eventType: "ADMIN_TRANSFERRED",
          },
        },
        create: {
          streamId: 0n,
          eventType: "ADMIN_TRANSFERRED",
          transactionHash: event.txHash,
          ledgerSequence: event.ledger,
          timestamp,
          metadata: JSON.stringify({
            previous_admin: previousAdmin,
            new_admin: newAdmin,
            transactionHash: event.txHash,
          }),
        },
        update: {
          ledgerSequence: event.ledger,
          timestamp,
        },
      });
    });

    sseService.broadcastToAdmin("stream.admin_transferred", {
      previousAdmin,
      newAdmin,
      transactionHash: event.txHash,
      ledger: event.ledger,
      timestamp,
    });
  }

  // ─── Event Handlers ────────────────────────────────────────────────────────

  private async handleStreamCreated(
    event: rpc.Api.EventResponse,
    streamIdTopic: xdr.ScVal,
  ): Promise<void> {
    const streamId = decodeU64(streamIdTopic);
    const body = decodeMap(event.value);

    if (
      !body["sender"] ||
      !body["recipient"] ||
      !body["token_address"] ||
      !body["rate_per_second"] ||
      !body["deposited_amount"] ||
      !body["start_time"]
    ) {
      throw new Error(`StreamCreated #${streamId}: missing body fields`);
    }

    const sender = decodeAddress(body["sender"]);
    const recipient = decodeAddress(body["recipient"]);
    const tokenAddress = decodeAddress(body["token_address"]);
    const ratePerSecond = decodeI128(body["rate_per_second"]);
    const depositedAmount = decodeI128(body["deposited_amount"]);
    const startTime = BigInt(decodeU64(body["start_time"]));

    const ratePerSecondBigInt = BigInt(ratePerSecond);
    const endTime =
      ratePerSecondBigInt === 0n
        ? null
        : startTime + BigInt(depositedAmount) / ratePerSecondBigInt;

    await prisma.$transaction(async (tx: Prisma.TransactionClient) => {
      await tx.user.upsert({
        where: { publicKey: sender },
        create: { publicKey: sender },
        update: {},
      });
      await tx.user.upsert({
        where: { publicKey: recipient },
        create: { publicKey: recipient },
        update: {},
      });

      await tx.stream.upsert({
        where: { streamId },
        create: {
          streamId,
          sender,
          recipient,
          tokenAddress,
          ratePerSecond,
          depositedAmount,
          withdrawnAmount: "0",
          startTime,
          endTime,
          lastUpdateTime: startTime,
          isActive: true,
        },
        update: {
          tokenAddress,
          ratePerSecond,
          depositedAmount,
          startTime,
          endTime,
          lastUpdateTime: startTime,
          isActive: true,
        },
      });

      const existingEvent = await tx.streamEvent.findUnique({
        where: {
          transactionHash_eventType: {
            transactionHash: event.txHash,
            eventType: "CREATED",
          },
        },
        select: { id: true },
      });
      if (existingEvent) {
        logger.warn(
          `[SorobanWorker] Duplicate StreamEvent skipped: txHash=${event.txHash} type=CREATED`,
        );
      } else {
        await tx.streamEvent.upsert({
          where: {
            transactionHash_eventType: {
              transactionHash: event.txHash,
              eventType: "CREATED",
            },
          },
          create: {
            streamId,
            eventType: "CREATED",
            amount: depositedAmount,
            transactionHash: event.txHash,
            ledgerSequence: event.ledger,
            timestamp: startTime,
            metadata: JSON.stringify({ tokenAddress, ratePerSecond }),
          },
          update: {
            ledgerSequence: event.ledger,
            timestamp: startTime,
          },
        });
      }
    });

    sseService.broadcastToStream(String(streamId), "stream.created", {
      streamId,
      sender,
      recipient,
      tokenAddress,
      ratePerSecond,
      depositedAmount,
      startTime,
      transactionHash: event.txHash,
      ledger: event.ledger,
    });
  }

  private async handleStreamToppedUp(
    event: rpc.Api.EventResponse,
    streamIdTopic: xdr.ScVal,
  ): Promise<void> {
    const streamId = decodeU64(streamIdTopic);
    const body = decodeMap(event.value);

    if (!body["amount"] || !body["new_deposited_amount"]) {
      throw new Error(`StreamToppedUp #${streamId}: missing body fields`);
    }

    const amount = decodeI128(body["amount"]);
    const newDepositedAmount = decodeI128(body["new_deposited_amount"]);
    const timestamp = Math.floor(Date.now() / 1000);

    await prisma.$transaction(async (tx: Prisma.TransactionClient) => {
      // Check for a duplicate BEFORE mutating any Stream fields so that a
      // replayed event never re-applies the top-up.
      const existingEvent = await tx.streamEvent.findUnique({
        where: { transactionHash_eventType: { transactionHash: event.txHash, eventType: 'TOPPED_UP' } },
        select: { id: true },
      });
      if (existingEvent) {
        logger.warn(`[SorobanWorker] Duplicate StreamEvent skipped: txHash=${event.txHash} type=TOPPED_UP`);
        return;
      }

      const stream = await tx.stream.findUniqueOrThrow({
        where: { streamId },
        select: {
          ratePerSecond: true,
          startTime: true,
          totalPausedDuration: true,
        },
      });

      const ratePerSecondBigInt = BigInt(stream.ratePerSecond);
      const newEndTime =
        ratePerSecondBigInt === 0n
          ? null
          : BigInt(stream.startTime) +
            (BigInt(newDepositedAmount) / ratePerSecondBigInt) +
            BigInt(stream.totalPausedDuration);

      await tx.stream.update({
        where: { streamId },
        data: {
          depositedAmount: newDepositedAmount,
          endTime: newEndTime,
          lastUpdateTime: timestamp,
        },
      });

      await tx.streamEvent.upsert({
        where: { transactionHash_eventType: { transactionHash: event.txHash, eventType: 'TOPPED_UP' } },
        create: {
          streamId,
          eventType: 'TOPPED_UP',
          amount,
          transactionHash: event.txHash,
          ledgerSequence: event.ledger,
          timestamp,
          metadata: JSON.stringify({ newDepositedAmount }),
        },
        update: {
          ledgerSequence: event.ledger,
          timestamp,
        },
      });
    });

    sseService.broadcastToStream(String(streamId), "stream.topped_up", {
      streamId,
      amount,
      newDepositedAmount,
      transactionHash: event.txHash,
      ledger: event.ledger,
      timestamp,
    });
  }

  private async handleTokensWithdrawn(
    event: rpc.Api.EventResponse,
    streamIdTopic: xdr.ScVal,
  ): Promise<void> {
    const streamId = decodeU64(streamIdTopic);
    const body = decodeMap(event.value);

    if (!body["recipient"] || !body["amount"] || !body["timestamp"]) {
      throw new Error(`TokensWithdrawn #${streamId}: missing body fields`);
    }

    const recipient = decodeAddress(body["recipient"]);
    const amount = decodeI128(body["amount"]);
    const timestamp = Number(decodeU64(body["timestamp"]));

    await prisma.$transaction(async (tx: Prisma.TransactionClient) => {
      // Check for a duplicate BEFORE mutating any Stream fields so that a
      // replayed event never double-increments withdrawnAmount.
      const existingEvent = await tx.streamEvent.findUnique({
        where: { transactionHash_eventType: { transactionHash: event.txHash, eventType: 'WITHDRAWN' } },
        select: { id: true },
      });
      if (existingEvent) {
        logger.warn(`[SorobanWorker] Duplicate StreamEvent skipped: txHash=${event.txHash} type=WITHDRAWN`);
        return;
      }

      const stream = await tx.stream.findUniqueOrThrow({
        where: { streamId },
        select: { withdrawnAmount: true },
      });

      const newWithdrawnAmount = (
        BigInt(stream.withdrawnAmount) + BigInt(amount)
      ).toString();

      await tx.stream.update({
        where: { streamId },
        data: {
          withdrawnAmount: newWithdrawnAmount,
          lastUpdateTime: timestamp,
        },
      });

      await tx.streamEvent.upsert({
        where: { transactionHash_eventType: { transactionHash: event.txHash, eventType: 'WITHDRAWN' } },
        create: {
          streamId,
          eventType: 'WITHDRAWN',
          amount,
          transactionHash: event.txHash,
          ledgerSequence: event.ledger,
          timestamp,
          metadata: JSON.stringify({ recipient }),
        },
        update: {
          ledgerSequence: event.ledger,
          timestamp,
        },
      });
    });

    sseService.broadcastToStream(String(streamId), "stream.withdrawn", {
      streamId,
      recipient,
      amount,
      transactionHash: event.txHash,
      ledger: event.ledger,
      timestamp,
    });
  }

  private async handleStreamCancelled(
    event: rpc.Api.EventResponse,
    streamIdTopic: xdr.ScVal,
  ): Promise<void> {
    const streamId = decodeU64(streamIdTopic);
    const body = decodeMap(event.value);

    if (!body["amount_withdrawn"] || !body["refunded_amount"]) {
      throw new Error(`StreamCancelled #${streamId}: missing body fields`);
    }

    const amountWithdrawn = decodeI128(body["amount_withdrawn"]);
    const refundedAmount = decodeI128(body["refunded_amount"]);
    const timestamp = Math.floor(Date.now() / 1000);

    await prisma.$transaction(async (tx: Prisma.TransactionClient) => {
      await tx.stream.update({
        where: { streamId },
        data: {
          isActive: false,
          withdrawnAmount: amountWithdrawn,
          lastUpdateTime: timestamp,
        },
      });

      const existingEvent = await tx.streamEvent.findUnique({
        where: {
          transactionHash_eventType: {
            transactionHash: event.txHash,
            eventType: "CANCELLED",
          },
        },
        select: { id: true },
      });
      if (existingEvent) {
        logger.warn(
          `[SorobanWorker] Duplicate StreamEvent skipped: txHash=${event.txHash} type=CANCELLED`,
        );
      } else {
        await tx.streamEvent.upsert({
          where: {
            transactionHash_eventType: {
              transactionHash: event.txHash,
              eventType: "CANCELLED",
            },
          },
          create: {
            streamId,
            eventType: "CANCELLED",
            amount: refundedAmount,
            transactionHash: event.txHash,
            ledgerSequence: event.ledger,
            timestamp,
            metadata: JSON.stringify({ amountWithdrawn, refundedAmount }),
          },
          update: {
            ledgerSequence: event.ledger,
            timestamp,
          },
        });
      }
    });

    sseService.broadcastToStream(String(streamId), "stream.cancelled", {
      streamId,
      refundedAmount,
      amountWithdrawn,
      transactionHash: event.txHash,
      ledger: event.ledger,
      timestamp,
    });
  }

  private async handleStreamCompleted(
    event: rpc.Api.EventResponse,
    streamIdTopic: xdr.ScVal,
  ): Promise<void> {
    const streamId = decodeU64(streamIdTopic);
    const body = decodeMap(event.value);

    if (!body["recipient"] || !body["total_withdrawn"]) {
      throw new Error(`StreamCompleted #${streamId}: missing body fields`);
    }

    const recipient = decodeAddress(body["recipient"]);
    const totalWithdrawn = decodeI128(body["total_withdrawn"]);
    const timestamp = Math.floor(Date.now() / 1000);

    await prisma.$transaction(async (tx: Prisma.TransactionClient) => {
      await tx.stream.update({
        where: { streamId },
        data: {
          isActive: false,
          withdrawnAmount: totalWithdrawn,
          lastUpdateTime: timestamp,
        },
      });

      const existingEvent = await tx.streamEvent.findUnique({
        where: {
          transactionHash_eventType: {
            transactionHash: event.txHash,
            eventType: "COMPLETED",
          },
        },
        select: { id: true },
      });
      if (existingEvent) {
        logger.warn(
          `[SorobanWorker] Duplicate StreamEvent skipped: txHash=${event.txHash} type=COMPLETED`,
        );
      } else {
        await tx.streamEvent.upsert({
          where: {
            transactionHash_eventType: {
              transactionHash: event.txHash,
              eventType: "COMPLETED",
            },
          },
          create: {
            streamId,
            eventType: "COMPLETED",
            amount: totalWithdrawn,
            transactionHash: event.txHash,
            ledgerSequence: event.ledger,
            timestamp,
            metadata: JSON.stringify({ recipient }),
          },
          update: {
            ledgerSequence: event.ledger,
            timestamp,
          },
        });
      }
    });

    sseService.broadcastToStream(String(streamId), "stream.completed", {
      streamId,
      recipient,
      totalWithdrawn,
      transactionHash: event.txHash,
      ledger: event.ledger,
      timestamp,
    });
  }

  private async handleFeeCollected(
    event: rpc.Api.EventResponse,
    streamIdTopic: xdr.ScVal,
  ): Promise<void> {
    const streamId = decodeU64(streamIdTopic);
    const body = decodeMap(event.value);

    if (!body["treasury"] || !body["fee_amount"] || !body["token"]) {
      throw new Error(`FeeCollected #${streamId}: missing body fields`);
    }

    const treasury = decodeAddress(body["treasury"]);
    const feeAmount = decodeI128(body["fee_amount"]);
    const token = decodeAddress(body["token"]);
    const timestamp = Math.floor(Date.now() / 1000);

    await prisma.$transaction(async (tx: Prisma.TransactionClient) => {
      const existingEvent = await tx.streamEvent.findUnique({
        where: {
          transactionHash_eventType: {
            transactionHash: event.txHash,
            eventType: "FEE_COLLECTED",
          },
        },
        select: { id: true },
      });
      if (existingEvent) {
        logger.warn(
          `[SorobanWorker] Duplicate StreamEvent skipped: txHash=${event.txHash} type=FEE_COLLECTED`,
        );
      } else {
        await tx.streamEvent.upsert({
          where: {
            transactionHash_eventType: {
              transactionHash: event.txHash,
              eventType: "FEE_COLLECTED",
            },
          },
          create: {
            streamId,
            eventType: "FEE_COLLECTED",
            amount: feeAmount,
            transactionHash: event.txHash,
            ledgerSequence: event.ledger,
            timestamp,
            metadata: JSON.stringify({ treasury, token }),
          },
          update: {},
        });
      }
    });

    // Broadcast to admin channel for treasury reporting
    sseService.broadcastToAdmin("stream.fee_collected", {
      streamId,
      treasury,
      feeAmount,
      token,
      transactionHash: event.txHash,
      ledger: event.ledger,
      timestamp,
    });
  }
  private async handleStreamPaused(
    event: rpc.Api.EventResponse,
    streamIdTopic: xdr.ScVal,
  ): Promise<void> {
    const streamId = decodeU64(streamIdTopic);
    const body = decodeMap(event.value);

    if (!body["sender"] || !body["paused_at"]) {
      throw new Error(`StreamPaused #${streamId}: missing body fields`);
    }

    const sender = decodeAddress(body["sender"]);
    const pausedAt = Number(decodeU64(body["paused_at"]));
    const timestamp = Math.floor(Date.now() / 1000);

    await prisma.$transaction(async (tx: Prisma.TransactionClient) => {
      await tx.stream.update({
        where: { streamId },
        data: {
          isPaused: true,
          pausedAt,
          lastUpdateTime: timestamp,
        },
      });

      const existingEvent = await tx.streamEvent.findUnique({
        where: {
          transactionHash_eventType: {
            transactionHash: event.txHash,
            eventType: "PAUSED",
          },
        },
        select: { id: true },
      });
      if (existingEvent) {
        logger.warn(
          `[SorobanWorker] Duplicate StreamEvent skipped: txHash=${event.txHash} type=PAUSED`,
        );
      } else {
        await tx.streamEvent.upsert({
          where: {
            transactionHash_eventType: {
              transactionHash: event.txHash,
              eventType: "PAUSED",
            },
          },
          create: {
            streamId,
            eventType: "PAUSED",
            transactionHash: event.txHash,
            ledgerSequence: event.ledger,
            timestamp,
            metadata: JSON.stringify({ sender, pausedAt }),
          },
          update: {
            ledgerSequence: event.ledger,
            timestamp,
          },
        });
      }
    });

    sseService.broadcastToStream(String(streamId), "stream.paused", {
      streamId,
      sender,
      pausedAt,
      transactionHash: event.txHash,
      ledger: event.ledger,
      timestamp,
    });
  }

  private async handleStreamResumed(
    event: rpc.Api.EventResponse,
    streamIdTopic: xdr.ScVal,
  ): Promise<void> {
    const streamId = decodeU64(streamIdTopic);
    const body = decodeMap(event.value);

    if (!body["sender"] || !body["new_end_time"]) {
      throw new Error(`StreamResumed #${streamId}: missing body fields`);
    }

    const sender = decodeAddress(body["sender"]);
    const newEndTime = decodeU64(body["new_end_time"]);
    const timestamp = Math.floor(Date.now() / 1000);

    await prisma.$transaction(async (tx: Prisma.TransactionClient) => {
      // Get current stream to calculate paused duration
      const currentStream = await tx.stream.findUniqueOrThrow({
        where: { streamId },
        select: { pausedAt: true, totalPausedDuration: true },
      });

      const additionalPausedDuration = currentStream.pausedAt
        ? timestamp - Number(currentStream.pausedAt)
        : 0;

      const newTotalPausedDuration =
        currentStream.totalPausedDuration + additionalPausedDuration;

      await tx.stream.update({
        where: { streamId },
        data: {
          isPaused: false,
          pausedAt: null,
          endTime: newEndTime,
          totalPausedDuration: newTotalPausedDuration,
          lastUpdateTime: timestamp,
        },
      });

      const existingEvent = await tx.streamEvent.findUnique({
        where: {
          transactionHash_eventType: {
            transactionHash: event.txHash,
            eventType: "RESUMED",
          },
        },
        select: { id: true },
      });
      if (existingEvent) {
        logger.warn(
          `[SorobanWorker] Duplicate StreamEvent skipped: txHash=${event.txHash} type=RESUMED`,
        );
      } else {
        await tx.streamEvent.upsert({
          where: {
            transactionHash_eventType: {
              transactionHash: event.txHash,
              eventType: "RESUMED",
            },
          },
          create: {
            streamId,
            eventType: "RESUMED",
            transactionHash: event.txHash,
            ledgerSequence: event.ledger,
            timestamp,
            metadata: JSON.stringify({
              sender,
              newEndTime: newEndTime.toString(),
              pausedDuration: additionalPausedDuration,
              totalPausedDuration: newTotalPausedDuration,
            }),
          },
          update: {
            ledgerSequence: event.ledger,
            timestamp,
          },
        });
      }
    });

    sseService.broadcastToStream(String(streamId), "stream.resumed", {
      streamId,
      sender,
      newEndTime: newEndTime.toString(),
      transactionHash: event.txHash,
      ledger: event.ledger,
      timestamp,
    });
  }
}

export const sorobanEventWorker = new SorobanEventWorker();