feat(replicas): auto-heal slave data drift and no-store public settings

Add sync-replica-data repair on deploy, a publisher-side replica watch
(60s + after tenant/settings writes), REFRESH PUBLICATION after migrate,
and Cache-Control no-store on program-settings and get_version.

Co-authored-by: Cursor <cursoragent@cursor.com>

src/app.module.ts

@@ -16,10 +16,12 @@ import { SupportModule } from './support/support.module';
 import { NotificationsModule } from './notifications/notifications.module';
 import { AdminsModule } from './admins/admins.module';
 import { HealthController } from './health.controller';
+import { ReplicaWatchModule } from './replica-watch/replica-watch.module';
 
 @Module({
   imports: [
     ConfigModule.forRoot({ isGlobal: true }),
+    ReplicaWatchModule,
     PrismaModule,
     NotificationsModule,
     AuthModule,

src/get-version/get-version.controller.ts

@@ -1,4 +1,4 @@
-import { Controller, Get, Headers, Query } from '@nestjs/common';
+import { Controller, Get, Header, Headers, Query } from '@nestjs/common';
 import { GetVersionService } from './get-version.service';
 
 @Controller()
@@ -6,6 +6,7 @@ export class GetVersionController {
   constructor(private readonly getVersionService: GetVersionService) {}
 
   @Get('get_version')
+  @Header('Cache-Control', 'no-store')
   getVersion(
     @Headers('host') host?: string,
     @Headers('x-forwarded-host') forwardedHost?: string,

src/program-settings/api-public-program-settings.controller.ts

@@ -6,7 +6,7 @@ export class ApiPublicProgramSettingsController {
   constructor(private readonly programSettings: ProgramSettingsService) {}
 
   @Get()
-  @Header('Cache-Control', 'public, max-age=30')
+  @Header('Cache-Control', 'no-store')
   getSettings(
     @Headers('host') host?: string,
     @Headers('x-forwarded-host') forwardedHost?: string,

src/program-settings/program-settings.service.ts

@@ -14,6 +14,7 @@ import { join } from 'path';
 import { Prisma, ProductType, Setting, SettingTemplate } from '@prisma/client';
 import { PrismaService } from '../prisma/prisma.service';
 import { AuditService } from '../audit/audit.service';
+import { ReplicaWatchService } from '../replica-watch/replica-watch.service';
 import { isSecretOrServerOnlySettingKey } from './secret-setting-keys';
 import { isRuntimeProgramSettingKey } from './runtime-setting-keys';
 import { DEPLOY_SETTING_CATALOG_KEYS } from './deploy-setting-catalog-keys';
@@ -57,6 +58,7 @@ export class ProgramSettingsService implements OnModuleInit {
   constructor(
     private readonly prisma: PrismaService,
     private readonly audit: AuditService,
+    private readonly replicaWatch: ReplicaWatchService,
   ) {}
 
   async onModuleInit() {
@@ -173,6 +175,7 @@ export class ProgramSettingsService implements OnModuleInit {
     });
 
     const items = await this.listTenantRows(tenantId);
+    this.replicaWatch.kick();
     return {
       tenant: this.tenantMeta(tenant),
       items,
@@ -320,6 +323,7 @@ export class ProgramSettingsService implements OnModuleInit {
     });
 
     const items = await this.listTenantRows(tenantId);
+    this.replicaWatch.kick();
     return {
       tenant: this.tenantMeta(tenant),
       apiBase,
@@ -390,6 +394,7 @@ export class ProgramSettingsService implements OnModuleInit {
       after: { keys: incoming.map((r) => r.key), tenants: tenants.length } as Prisma.InputJsonValue,
     });
 
+    this.replicaWatch.kick();
     return {
       keys: incoming.map((r) => r.key),
       tenantsUpdated: tenants.length,
@@ -416,6 +421,7 @@ export class ProgramSettingsService implements OnModuleInit {
       after: result as Prisma.InputJsonValue,
     });
 
+    this.replicaWatch.kick();
     return { key, ...result };
   }
 

src/program-settings/public-program-settings.controller.ts

@@ -6,7 +6,7 @@ export class PublicProgramSettingsController {
   constructor(private readonly programSettings: ProgramSettingsService) {}
 
   @Get('get_settings')
-  @Header('Cache-Control', 'public, max-age=30')
+  @Header('Cache-Control', 'no-store')
   getSettings(
     @Headers('host') host?: string,
     @Headers('x-forwarded-host') forwardedHost?: string,

src/replica-watch/replica-watch.module.ts

@@ -0,0 +1,9 @@
+import { Global, Module } from '@nestjs/common';
+import { ReplicaWatchService } from './replica-watch.service';
+
+@Global()
+@Module({
+  providers: [ReplicaWatchService],
+  exports: [ReplicaWatchService],
+})
+export class ReplicaWatchModule {}

src/replica-watch/replica-watch.service.ts

@@ -0,0 +1,93 @@
+import { Injectable, Logger, OnModuleDestroy, OnModuleInit } from '@nestjs/common';
+import { spawn } from 'child_process';
+import { existsSync } from 'fs';
+import { join } from 'path';
+
+/**
+ * Logical replication streams WAL, it does not re-copy missed history.
+ * If a replica drifted (new table never REFRESH'd, apply error, …) this
+ * process heals it on the publisher — no operator command required.
+ */
+@Injectable()
+export class ReplicaWatchService implements OnModuleInit, OnModuleDestroy {
+  private readonly logger = new Logger(ReplicaWatchService.name);
+  private timer?: ReturnType<typeof setInterval>;
+  private running = false;
+  private pending = false;
+
+  onModuleInit() {
+    if (!this.enabled()) {
+      this.logger.log('replica watch idle (no deploy/replicas.env on this host)');
+      return;
+    }
+    this.logger.log('replica watch on — drift auto-repairs every 60s');
+    this.timer = setInterval(() => this.kick(), 60_000);
+    setTimeout(() => this.kick(), 8_000);
+  }
+
+  onModuleDestroy() {
+    if (this.timer) clearInterval(this.timer);
+  }
+
+  kick() {
+    if (!this.enabled()) return;
+    if (this.running) {
+      this.pending = true;
+      return;
+    }
+    void this.run();
+  }
+
+  private enabled() {
+    return existsSync(this.envFile()) && existsSync(this.scriptPath());
+  }
+
+  private envFile() {
+    return join(process.cwd(), 'deploy/replicas.env');
+  }
+
+  private scriptPath() {
+    return join(process.cwd(), 'deploy/sync-replica-data.sh');
+  }
+
+  private run(): Promise<void> {
+    this.running = true;
+    return new Promise((resolve) => {
+      const child = spawn('bash', [this.scriptPath(), '--repair-if-needed'], {
+        cwd: process.cwd(),
+        env: process.env,
+      });
+      let out = '';
+      child.stdout?.on('data', (buf: Buffer) => {
+        out += buf.toString();
+      });
+      child.stderr?.on('data', (buf: Buffer) => {
+        out += buf.toString();
+      });
+      child.on('error', (err) => {
+        this.logger.warn(`replica watch failed to start: ${err.message}`);
+        this.finish();
+        resolve();
+      });
+      child.on('close', (code) => {
+        const text = out.trim();
+        const healed = /Drift detected|repair done|Replica data sync complete/i.test(text);
+        if (code !== 0) {
+          this.logger.warn(`replica watch exit ${code}\n${text.slice(-2000)}`);
+        } else if (healed) {
+          this.logger.log(text.slice(-2000) || 'replica data repaired');
+        }
+        this.finish();
+        resolve();
+      });
+    });
+  }
+
+  private finish() {
+    this.running = false;
+    if (this.pending) {
+      this.pending = false;
+      void this.run();
+    }
+  }
+}

src/tenants/tenants.service.ts

@@ -8,6 +8,7 @@ import { PrismaService } from '../prisma/prisma.service';
 import { AuditService } from '../audit/audit.service';
 import { AdminNotifyService } from '../notifications/admin-notify.service';
 import { ProgramSettingsService } from '../program-settings/program-settings.service';
+import { ReplicaWatchService } from '../replica-watch/replica-watch.service';
 import { CreateTenantDto } from './dto/create-tenant.dto';
 import { UpdateTenantDto } from './dto/update-tenant.dto';
 
@@ -18,6 +19,7 @@ export class TenantsService {
     private readonly audit: AuditService,
     private readonly notify: AdminNotifyService,
     private readonly programSettings: ProgramSettingsService,
+    private readonly replicaWatch: ReplicaWatchService,
   ) {}
 
   async findAll(product?: ProductType) {
@@ -124,6 +126,7 @@ export class TenantsService {
         `سوپرادمین: نسخه جدید «${tenant.nameFa}» (${tenant.domain}) ایجاد شد.`,
       );
 
+      this.replicaWatch.kick();
       return this.sanitize(tenant);
     } catch (error) {
       this.handleUnique(error);
@@ -245,6 +248,7 @@ export class TenantsService {
           : `سوپرادمین: نسخه «${tenant.nameFa}» به‌روزرسانی شد.`,
       );
 
+      this.replicaWatch.kick();
       return this.sanitize(tenant);
     } catch (error) {
       this.handleUnique(error);
@@ -273,6 +277,7 @@ export class TenantsService {
       `سوپرادمین: نسخه «${before.nameFa}» (${before.domain}) حذف شد.`,
     );
 
+    this.replicaWatch.kick();
     return { ok: true };
   }