feat(sync): add sync subsystem: SyncManager, OpsServer sync handlers, Sync UI and state, provider groupFilter support, and realtime sync log streaming via TypedSocket

ts/opsserver/classes.opsserver.ts

@@ -20,6 +20,7 @@ export class OpsServer {
   public webhookHandler!: handlers.WebhookHandler;
   public actionsHandler!: handlers.ActionsHandler;
   public actionLogHandler!: handlers.ActionLogHandler;
+  public syncHandler!: handlers.SyncHandler;
 
   constructor(gitopsAppRef: GitopsApp) {
     this.gitopsAppRef = gitopsAppRef;
@@ -63,6 +64,7 @@ export class OpsServer {
     this.logsHandler = new handlers.LogsHandler(this);
     this.actionsHandler = new handlers.ActionsHandler(this);
     this.actionLogHandler = new handlers.ActionLogHandler(this);
+    this.syncHandler = new handlers.SyncHandler(this);
 
     logger.success('OpsServer TypedRequest handlers initialized');
   }

ts/opsserver/handlers/connections.handler.ts

@@ -39,6 +39,7 @@ export class ConnectionsHandler {
             dataArg.providerType,
             dataArg.baseUrl,
             dataArg.token,
+            dataArg.groupFilter,
           );
           this.actionLog.append({
             actionType: 'create',
@@ -65,12 +66,14 @@ export class ConnectionsHandler {
               name: dataArg.name,
               baseUrl: dataArg.baseUrl,
               token: dataArg.token,
+              groupFilter: dataArg.groupFilter,
             },
           );
           const fields = [
             dataArg.name && 'name',
             dataArg.baseUrl && 'baseUrl',
             dataArg.token && 'token',
+            dataArg.groupFilter !== undefined && 'groupFilter',
           ].filter(Boolean).join(', ');
           this.actionLog.append({
             actionType: 'update',

ts/opsserver/handlers/index.ts

@@ -8,3 +8,4 @@ export { LogsHandler } from './logs.handler.ts';
 export { WebhookHandler } from './webhook.handler.ts';
 export { ActionsHandler } from './actions.handler.ts';
 export { ActionLogHandler } from './actionlog.handler.ts';
+export { SyncHandler } from './sync.handler.ts';

ts/opsserver/handlers/sync.handler.ts

@@ -0,0 +1,222 @@
+import * as plugins from '../../plugins.ts';
+import type { OpsServer } from '../classes.opsserver.ts';
+import * as interfaces from '../../../ts_interfaces/index.ts';
+import { requireValidIdentity } from '../helpers/guards.ts';
+import { logger } from '../../logging.ts';
+
+export class SyncHandler {
+  public typedrouter = new plugins.typedrequest.TypedRouter();
+
+  constructor(private opsServerRef: OpsServer) {
+    this.opsServerRef.typedrouter.addTypedRouter(this.typedrouter);
+    this.registerHandlers();
+    this.setupBroadcast();
+  }
+
+  /**
+   * Wire up the logger's broadcast function to push sync log entries
+   * to all connected frontends via TypedSocket.
+   */
+  private setupBroadcast(): void {
+    logger.setBroadcastFn((entry) => {
+      try {
+        const typedsocket = this.opsServerRef.server?.typedserver?.typedsocket;
+        if (!typedsocket) return;
+        typedsocket.findAllTargetConnectionsByTag('allClients').then((connections) => {
+          for (const conn of connections) {
+            typedsocket
+              .createTypedRequest<interfaces.requests.IReq_PushSyncLog>('pushSyncLog', conn)
+              .fire({ entry })
+              .catch(() => {});
+          }
+        }).catch(() => {});
+      } catch {
+        // Server may not be ready yet — ignore
+      }
+    });
+  }
+
+  private get syncManager() {
+    return this.opsServerRef.gitopsAppRef.syncManager;
+  }
+
+  private get actionLog() {
+    return this.opsServerRef.gitopsAppRef.actionLog;
+  }
+
+  private registerHandlers(): void {
+    // Get all sync configs
+    this.typedrouter.addTypedHandler(
+      new plugins.typedrequest.TypedHandler<interfaces.requests.IReq_GetSyncConfigs>(
+        'getSyncConfigs',
+        async (dataArg) => {
+          await requireValidIdentity(this.opsServerRef.adminHandler, dataArg);
+          return { configs: this.syncManager.getConfigs() };
+        },
+      ),
+    );
+
+    // Create sync config
+    this.typedrouter.addTypedHandler(
+      new plugins.typedrequest.TypedHandler<interfaces.requests.IReq_CreateSyncConfig>(
+        'createSyncConfig',
+        async (dataArg) => {
+          await requireValidIdentity(this.opsServerRef.adminHandler, dataArg);
+          const config = await this.syncManager.createConfig({
+            name: dataArg.name,
+            sourceConnectionId: dataArg.sourceConnectionId,
+            targetConnectionId: dataArg.targetConnectionId,
+            targetGroupOffset: dataArg.targetGroupOffset,
+            intervalMinutes: dataArg.intervalMinutes,
+            enforceDelete: dataArg.enforceDelete,
+            enforceGroupDelete: dataArg.enforceGroupDelete,
+            addMirrorHint: dataArg.addMirrorHint,
+          });
+          this.actionLog.append({
+            actionType: 'create',
+            entityType: 'sync',
+            entityId: config.id,
+            entityName: config.name,
+            details: `Created sync config "${config.name}" (${config.intervalMinutes}m interval)`,
+            username: dataArg.identity.username,
+          });
+          return { config };
+        },
+      ),
+    );
+
+    // Update sync config
+    this.typedrouter.addTypedHandler(
+      new plugins.typedrequest.TypedHandler<interfaces.requests.IReq_UpdateSyncConfig>(
+        'updateSyncConfig',
+        async (dataArg) => {
+          await requireValidIdentity(this.opsServerRef.adminHandler, dataArg);
+          const config = await this.syncManager.updateConfig(dataArg.syncConfigId, {
+            name: dataArg.name,
+            targetGroupOffset: dataArg.targetGroupOffset,
+            intervalMinutes: dataArg.intervalMinutes,
+            enforceDelete: dataArg.enforceDelete,
+            enforceGroupDelete: dataArg.enforceGroupDelete,
+            addMirrorHint: dataArg.addMirrorHint,
+          });
+          this.actionLog.append({
+            actionType: 'update',
+            entityType: 'sync',
+            entityId: config.id,
+            entityName: config.name,
+            details: `Updated sync config "${config.name}"`,
+            username: dataArg.identity.username,
+          });
+          return { config };
+        },
+      ),
+    );
+
+    // Delete sync config
+    this.typedrouter.addTypedHandler(
+      new plugins.typedrequest.TypedHandler<interfaces.requests.IReq_DeleteSyncConfig>(
+        'deleteSyncConfig',
+        async (dataArg) => {
+          await requireValidIdentity(this.opsServerRef.adminHandler, dataArg);
+          const config = this.syncManager.getConfig(dataArg.syncConfigId);
+          await this.syncManager.deleteConfig(dataArg.syncConfigId);
+          this.actionLog.append({
+            actionType: 'delete',
+            entityType: 'sync',
+            entityId: dataArg.syncConfigId,
+            entityName: config?.name || dataArg.syncConfigId,
+            details: `Deleted sync config "${config?.name || dataArg.syncConfigId}"`,
+            username: dataArg.identity.username,
+          });
+          return { ok: true };
+        },
+      ),
+    );
+
+    // Pause/resume sync config
+    this.typedrouter.addTypedHandler(
+      new plugins.typedrequest.TypedHandler<interfaces.requests.IReq_PauseSyncConfig>(
+        'pauseSyncConfig',
+        async (dataArg) => {
+          await requireValidIdentity(this.opsServerRef.adminHandler, dataArg);
+          const config = await this.syncManager.pauseConfig(
+            dataArg.syncConfigId,
+            dataArg.paused,
+          );
+          this.actionLog.append({
+            actionType: dataArg.paused ? 'pause' : 'resume',
+            entityType: 'sync',
+            entityId: config.id,
+            entityName: config.name,
+            details: `${dataArg.paused ? 'Paused' : 'Resumed'} sync config "${config.name}"`,
+            username: dataArg.identity.username,
+          });
+          return { config };
+        },
+      ),
+    );
+
+    // Trigger sync manually
+    this.typedrouter.addTypedHandler(
+      new plugins.typedrequest.TypedHandler<interfaces.requests.IReq_TriggerSync>(
+        'triggerSync',
+        async (dataArg) => {
+          await requireValidIdentity(this.opsServerRef.adminHandler, dataArg);
+          const config = this.syncManager.getConfig(dataArg.syncConfigId);
+          if (!config) {
+            return { ok: false, message: 'Sync config not found' };
+          }
+          // Fire and forget — force=true bypasses paused check for manual triggers
+          this.syncManager.executeSync(dataArg.syncConfigId, true).catch((err) => {
+            console.error(`Manual sync trigger failed: ${err}`);
+          });
+          this.actionLog.append({
+            actionType: 'sync',
+            entityType: 'sync',
+            entityId: config.id,
+            entityName: config.name,
+            details: `Manually triggered sync "${config.name}"`,
+            username: dataArg.identity.username,
+          });
+          return { ok: true, message: 'Sync triggered' };
+        },
+      ),
+    );
+
+    // Preview sync (dry run — shows source → target mappings)
+    this.typedrouter.addTypedHandler(
+      new plugins.typedrequest.TypedHandler<interfaces.requests.IReq_PreviewSync>(
+        'previewSync',
+        async (dataArg) => {
+          await requireValidIdentity(this.opsServerRef.adminHandler, dataArg);
+          const result = await this.syncManager.previewSync(dataArg.syncConfigId);
+          return { mappings: result.mappings, deletions: result.deletions, groupDeletions: result.groupDeletions };
+        },
+      ),
+    );
+
+    // Get repo statuses for a sync config
+    this.typedrouter.addTypedHandler(
+      new plugins.typedrequest.TypedHandler<interfaces.requests.IReq_GetSyncRepoStatuses>(
+        'getSyncRepoStatuses',
+        async (dataArg) => {
+          await requireValidIdentity(this.opsServerRef.adminHandler, dataArg);
+          const statuses = await this.syncManager.getRepoStatuses(dataArg.syncConfigId);
+          return { statuses };
+        },
+      ),
+    );
+
+    // Get recent sync log entries
+    this.typedrouter.addTypedHandler(
+      new plugins.typedrequest.TypedHandler<interfaces.requests.IReq_GetSyncLogs>(
+        'getSyncLogs',
+        async (dataArg) => {
+          await requireValidIdentity(this.opsServerRef.adminHandler, dataArg);
+          const logs = logger.getSyncLogs(dataArg.limit || 200);
+          return { logs };
+        },
+      ),
+    );
+  }
+}