Persist notifications and delivery attempts & Enforce payout lifecycle and authorization

app/backend/src/auth/decorators/roles.decorator.ts

@@ -0,0 +1,5 @@
+import { SetMetadata } from "@nestjs/common";
+import { UserRole } from "../enums/user-role.enum";
+
+export const ROLES_KEY = "roles";
+export const Roles = (...roles: UserRole[]) => SetMetadata(ROLES_KEY, roles);

app/backend/src/auth/enums/user-role.enum.ts

@@ -0,0 +1,4 @@
+export enum UserRole {
+  Admin = "admin",
+  User = "user",
+}

app/backend/src/auth/guards/roles.guard.ts

@@ -0,0 +1,21 @@
+import { Injectable, CanActivate, ExecutionContext } from "@nestjs/common";
+import { Reflector } from "@nestjs/core";
+import { ROLES_KEY } from "./roles.decorator";
+import { UserRole } from "../enums/user-role.enum";
+
+@Injectable()
+export class RolesGuard implements CanActivate {
+  constructor(private reflector: Reflector) {}
+
+  canActivate(context: ExecutionContext): boolean {
+    const requiredRoles = this.reflector.getAllAndOverride<UserRole[]>(
+      ROLES_KEY,
+      [context.getHandler(), context.getClass()],
+    );
+    if (!requiredRoles) {
+      return true;
+    }
+    const { user } = context.switchToHttp().getRequest();
+    return requiredRoles.some((role) => user.role?.includes(role));
+  }
+}

app/backend/src/common/errors/contract-adapter.error.ts

@@ -0,0 +1,14 @@
+export enum ContractAdapterErrorCode {
+  StreamError = "STREAM_ERROR",
+  ParseError = "PARSE_ERROR",
+  UnknownError = "UNKNOWN_ERROR",
+}
+
+export class ContractAdapterError extends Error {
+  constructor(
+    public readonly code: ContractAdapterErrorCode,
+    public readonly message: string,
+  ) {
+    super(message);
+  }
+}

app/backend/src/health/health.module.ts

@@ -1,14 +1,20 @@
 import { Module } from "@nestjs/common";
+import { HealthController } from "./health.controller";
+import { HealthService } from "./health.service";
 import { SupabaseModule } from "../supabase/supabase.module";
 import { StellarModule } from "../stellar/stellar.module";
 import { JobQueueModule } from "../job-queue/job-queue.module";
 import { IngestionModule } from "../ingestion/ingestion.module";
 import { TransactionsModule } from "../transactions/transactions.module";
-import { HealthController } from "./health.controller";
-import { HealthService } from "./health.service";
 
 @Module({
-  imports: [SupabaseModule, StellarModule, JobQueueModule, IngestionModule, TransactionsModule],
+  imports: [
+    SupabaseModule,
+    StellarModule,
+    JobQueueModule,
+    IngestionModule,
+    TransactionsModule,
+  ],
   controllers: [HealthController],
   providers: [HealthService],
 })

app/backend/src/health/health.service.ts

@@ -7,6 +7,7 @@ import { JobQueueService } from "../job-queue/job-queue.service";
 import { JobRepository } from "../job-queue/job.repository";
 import { CursorRepository } from "../ingestion/cursor.repository";
 import { SorobanRpcService } from "../transactions/soroban-rpc.service";
+import { StellarIngestionService } from "../ingestion/stellar-ingestion.service";
 
 @Injectable()
 export class HealthService {
@@ -22,6 +23,7 @@ export class HealthService {
     private readonly jobRepository: JobRepository,
     private readonly cursorRepository: CursorRepository,
     private readonly sorobanRpcService: SorobanRpcService,
+    private readonly stellarIngestionService: StellarIngestionService,
   ) {}
 
   /**
@@ -280,6 +282,33 @@ export class HealthService {
     }
   }
 
+  async checkStellarIngestion(): Promise<{
+    status: "up" | "down";
+    details?: string;
+  }> {
+    const { isRunning, contractId } = this.stellarIngestionService.getStatus();
+
+    if (!isRunning) {
+      return {
+        status: "down",
+        details: "Stellar ingestion service is not running",
+      };
+    }
+
+    if (!contractId) {
+      return {
+        status: "down",
+        details:
+          "Stellar ingestion service is not configured with a contract ID",
+      };
+    }
+
+    return {
+      status: "up",
+      details: `Stellar ingestion service is running for contract ${contractId}`,
+    };
+  }
+
   /**
    * Checks if database migrations are applied by querying the schema_migrations table.
    * This is a Supabase/PostgreSQL specific check.
@@ -348,19 +377,34 @@ export class HealthService {
    * Performs deep dependency checks for /ready.
    */
   async getReadinessStatus() {
-    const [supabase, env, migrations, queue, horizon, sorobanRpc, ingestion] =
-      await Promise.all([
-        this.checkSupabase(),
-        Promise.resolve(this.checkEnvironment()),
-        this.checkMigrations(),
-        this.checkQueue(),
-        this.checkHorizon(),
-        this.checkSorobanRpc(),
-        this.checkIngestionLag(),
-      ]);
+    const [
+      supabase,
+      env,
+      migrations,
+      queue,
+      horizon,
+      sorobanRpc,
+      ingestion,
+      stellarIngestion,
+    ] = await Promise.all([
+      this.checkSupabase(),
+      Promise.resolve(this.checkEnvironment()),
+      this.checkMigrations(),
+      this.checkQueue(),
+      this.checkHorizon(),
+      this.checkSorobanRpc(),
+      this.checkIngestionLag(),
+      this.checkStellarIngestion(),
+    ]);
 
     // Critical dependencies: database, migrations, queue, horizon
-    const criticalChecks = [supabase, migrations, queue, horizon];
+    const criticalChecks = [
+      supabase,
+      migrations,
+      queue,
+      horizon,
+      stellarIngestion,
+    ];
     const ready = criticalChecks.every((check) => check.status === "up");
 
     return {
@@ -415,6 +459,15 @@ export class HealthService {
           lastSuccess: ingestion.lastSuccess,
           error: ingestion.status === "down" ? ingestion.details : undefined,
         },
+        {
+          name: "stellar_ingestion",
+          status: stellarIngestion.status,
+          details: stellarIngestion.details,
+          error:
+            stellarIngestion.status === "down"
+              ? stellarIngestion.details
+              : undefined,
+        },
       ],
     };
   }

app/backend/src/ingestion/stellar-ingestion.service.ts

@@ -13,6 +13,10 @@ import {
   SorobanEventParser,
   RawHorizonContractEvent,
 } from "./soroban-event.parser";
+import {
+  ContractAdapterError,
+  ContractAdapterErrorCode,
+} from "../common/errors/contract-adapter.error";
 import { CursorRepository } from "./cursor.repository";
 import { EscrowEventRepository } from "./escrow-event.repository";
 import { JobQueueService } from "../job-queue/job-queue.service";
@@ -172,10 +176,15 @@ export class StellarIngestionService implements OnModuleInit, OnModuleDestroy {
       onmessage: (record: unknown) => {
         void this.handleRecord(record as RawHorizonContractEvent, streamId);
       },
+
       onerror: (err: unknown) => {
         this.logger.error(`Stream error for ${streamId}: ${String(err)}`);
         this.stopCurrentStream();
         this.scheduleReconnect(contractId);
+        throw new ContractAdapterError(
+          ContractAdapterErrorCode.StreamError,
+          String(err),
+        );
       },
     }) as () => void;
 
@@ -217,6 +226,10 @@ export class StellarIngestionService implements OnModuleInit, OnModuleDestroy {
       this.logger.error(`SSE error for ${streamId}: ${String(err)}`);
       es.close();
       this.scheduleReconnect(contractId);
+      throw new ContractAdapterError(
+        ContractAdapterErrorCode.StreamError,
+        String(err),
+      );
     };
 
     return () => es.close();
@@ -283,6 +296,10 @@ export class StellarIngestionService implements OnModuleInit, OnModuleDestroy {
         this.logger.error(`Stream error for ${streamId}: ${String(err)}`);
         this.stopCurrentStream();
         this.scheduleReconnect(contractId);
+        throw new ContractAdapterError(
+          ContractAdapterErrorCode.StreamError,
+          String(err),
+        );
       },
     }) as () => void;
 
@@ -365,6 +382,17 @@ export class StellarIngestionService implements OnModuleInit, OnModuleDestroy {
     }
   }
 
+  // ---------------------------------------------------------------------------
+  // Public status
+  // ---------------------------------------------------------------------------
+
+  getStatus(): { isRunning: boolean; contractId: string | null } {
+    return {
+      isRunning: !!this.stopStream,
+      contractId: this.currentContractId,
+    };
+  }
+
   // ---------------------------------------------------------------------------
   // Event processing
   // ---------------------------------------------------------------------------
@@ -373,23 +401,30 @@ export class StellarIngestionService implements OnModuleInit, OnModuleDestroy {
     raw: RawHorizonContractEvent,
     streamId: string,
   ): Promise<void> {
-    const event = this.parser.parse(raw);
+    try {
+      const event = this.parser.parse(raw);
 
-    if (!event) {
-      // Unrecognised or non- RustAcademy event; still advance cursor.
-      await this.safeUpdateCursor(streamId, raw.paging_token, raw.ledger);
-      return;
-    }
+      if (!event) {
+        // Unrecognised or non- RustAcademy event; still advance cursor.
+        await this.safeUpdateCursor(streamId, raw.paging_token, raw.ledger);
+        return;
+      }
 
-    this.logger.debug(
-      `Processing ${event.eventType} paging_token=${event.pagingToken}`,
-    );
+      this.logger.debug(
+        `Processing ${event.eventType} paging_token=${event.pagingToken}`,
+      );
 
-    await this.persistEvent(event);
-    await this.safeUpdateCursor(streamId, raw.paging_token, raw.ledger);
+      await this.persistEvent(event);
+      await this.safeUpdateCursor(streamId, raw.paging_token, raw.ledger);
 
-    // Emit for other services / notification layer
-    this.eventEmitter.emit(`stellar.${event.eventType}`, event);
+      // Emit for other services / notification layer
+      this.eventEmitter.emit(`stellar.${event.eventType}`, event);
+    } catch (err) {
+      throw new ContractAdapterError(
+        ContractAdapterErrorCode.ParseError,
+        `Failed to parse event: ${String(err)}`,
+      );
+    }
   }
 
   private async persistEvent(event: RustAcademyContractEvent): Promise<void> {

app/backend/src/payments/dto/payout.dto.ts

@@ -0,0 +1,14 @@
+import { ApiProperty } from "@nestjs/swagger";
+import { IsNotEmpty, IsNumber, IsString } from "class-validator";
+
+export class PayoutDto {
+  @ApiProperty()
+  @IsNotEmpty()
+  @IsString()
+  destinationAddress: string;
+
+  @ApiProperty()
+  @IsNotEmpty()
+  @IsNumber()
+  amount: number;
+}

app/backend/src/payments/entities/payout.entity.ts

@@ -0,0 +1,38 @@
+import {
+  Entity,
+  PrimaryGeneratedColumn,
+  Column,
+  CreateDateColumn,
+  UpdateDateColumn,
+} from "typeorm";
+
+export enum PayoutStatus {
+  Pending = "pending",
+  Released = "released",
+  Failed = "failed",
+}
+
+@Entity()
+export class Payout {
+  @PrimaryGeneratedColumn("uuid")
+  id: string;
+
+  @Column()
+  destinationAddress: string;
+
+  @Column()
+  amount: number;
+
+  @Column({
+    type: "enum",
+    enum: PayoutStatus,
+    default: PayoutStatus.Pending,
+  })
+  status: PayoutStatus;
+
+  @CreateDateColumn()
+  createdAt: Date;
+
+  @UpdateDateColumn()
+  updatedAt: Date;
+}