light weight scheduler and outbox sweeper (part 1)

Author: pramodnarayana

Makefile

@@ -57,7 +57,7 @@ dev-web:
 
 dev-worker-orchestrator:
 	@echo "Starting Orchestrator Worker for local development..."
-	ENVIRONMENT=development PYTHONPATH=services/workers/orchestrator/src:libs/database/src:libs/config/src:libs/pipeline/src:libs/domain/src:libs/transformer/src uv run python services/workers/orchestrator/src/worker/main.py
+	ENVIRONMENT=development PYTHONPATH=services/workers/orchestrator/src:libs/database/src:libs/config/src:libs/pipeline/src:libs/domain/src:libs/transformer/src:libs/scheduler/src uv run python services/workers/orchestrator/src/worker/main.py
 
 dev-worker-compute:
 	@echo "Starting Compute Worker for local development..."
@@ -92,7 +92,7 @@ db-sqs-reset: db-reset sqs-purge
 
 clear-data:
 	@echo "Clearing data plane tables (edi_message, edi_json, api_gateway, outbox) and purging SQS..."
-	uv run python scripts/clear_data.py
+	uv run python scripts/clear_data.py --i-am-sure
 
 seed: db-init
 

TECHNICAL_DEBT.md

@@ -22,16 +22,17 @@ To achieve the long-term vision of completely merging and modernizing `bots_core
 ## 2. Outbox Sweeper (CDC Fallback Relay)
 
 **Description:**
-The system currently relies exclusively on Debezium (CDC) reading the PostgreSQL Write-Ahead Log (WAL) to route `Outbox` events to SQS. If Debezium crashes, loses offsets, or experiences network partitioning, `PENDING` outbox events will be permanently trapped in the database, breaking the asynchronous event pipeline.
+The Outbox Sweeper has been implemented as a fallback to Debezium (CDC). It iterates over all shards to relay `PENDING` outbox events to SQS. However, the current implementation iterates over shards sequentially (`for shard in shards: await self._sweep_shard(...)`). As the number of shards grows in the multi-tenant architecture, this sequential sweep will take longer and potentially exceed the polling interval, causing lag.
 
 **Proposed Resolution:**
-Implement an Outbox Sweeper background worker that acts as a robust enterprise fallback and garbage collector:
-1. **Fallback Poller:** A cron/scheduled task that periodically queries `SELECT * FROM outbox WHERE status = 'PENDING'` for events older than a configured threshold (e.g., 60 seconds) and manually relays them to SQS.
-2. **Garbage Collector:** A cleanup task that runs `DELETE FROM outbox WHERE status = 'COMPLETED'` for events older than 7 days to prevent unbounded database growth.
+Refactor the Outbox Sweeper to use Bounded Concurrency or Distributed Job Fan-out:
+1. **Bounded Concurrency:** Run sweeps concurrently using `asyncio.gather` bounded by an `asyncio.Semaphore` so multiple shards are swept at once without exhausting resources.
+2. **Distributed Job Fan-out:** Instead of a single job, spawn a `ScheduledJob` for each shard dynamically, allowing multiple orchestrator pods to load balance the shard sweeping.
+3. **Garbage Collector (Pending):** A cleanup task that runs `DELETE FROM outbox WHERE status = 'COMPLETED'` for events older than 7 days to prevent unbounded database growth is still needed.
 
-**Estimated Effort:** Low
+**Estimated Effort:** Low-Medium
 **Estimated Time:** 1 to 2 days
-**Impact:** Essential for enterprise-grade high availability. Guarantees no messages are ever lost due to CDC infrastructure failures and keeps the database optimized over time.
+**Impact:** Prevents the sweeper from falling behind as the number of database shards scales, ensuring enterprise-grade multi-tenant reliability.
 
 ## 3. AS2 Protocol
 
@@ -80,3 +81,19 @@ Currently, the bots engine does not support a lightweight validation mode (e.g.,
 ### UnitOfWork Architecture (Control Plane vs Data Plane Naming)
 Currently, the `UnitOfWork` (and its underlying SQL Alchemy repositories) leak infrastructure/deployment boundaries ("Control Plane" and "Data Plane") into domain business logic. We have giant God-objects like `SqlAlchemyControlPlaneRepository` inheriting from 10+ distinct repositories, causing namespace collisions and violating SOLID principles (Single Responsibility Principle).
 **Future Action:** Refactor `UnitOfWork` to remove `control_plane` and `data_plane` concepts from class names and properties. Use Composition to expose distinct Bounded Contexts (e.g., `self.trading_partners`, `self.transactions`, `self.routes`) instead of lumping them into control/data plane buckets.
+
+## 8. Hybrid SQS Tenancy (Dynamic Queue Resolution)
+
+**Description:**
+Currently, both inbound and outbound events are routed to a static, shared SQS queue (e.g., `TransformOrchestrationQueue`). In a multi-tenant environment, a massive batch of outbound events from one tenant can block critical inbound processing for all other tenants (the "noisy neighbor" problem).
+
+**Proposed Resolution:**
+Implement a Hybrid SQS routing model that dynamically resolves the target queue based on the tenant's tier and the event direction:
+1. **Dynamic Queue Resolver:** The CDC Relay and Outbox Sweeper should read the `tenant_id` from the outbox event and lookup the tenant's tier (cached in memory).
+2. **Standard Tenants:** Route inbound events to `standard-inbound-queue` and outbound events to `standard-outbound-queue`.
+3. **Enterprise Tenants:** Route events to dedicated queues (e.g., `enterprise-{tenant_name}-inbound-queue`).
+4. **Dedicated Workers:** Deploy separate worker pods for standard inbound, standard outbound, and dedicated enterprise queues to provide strict compute isolation.
+
+**Estimated Effort:** Medium
+**Estimated Time:** 3 to 5 days
+**Impact:** Essential for enterprise-grade SaaS scaling. Guarantees compute isolation for enterprise customers and bulkheads heavy outbound processing from blocking high-priority inbound traffic.

docker/localstack/init-aws.sh

@@ -14,7 +14,11 @@ awslocal sqs create-queue --queue-name CDC-DLQ
 
 awslocal sqs create-queue --queue-name TransformOrchestrationQueue-DLQ
 TRANSFORM_DLQ_ARN=$(awslocal sqs get-queue-attributes --queue-url http://localhost:4566/000000000000/TransformOrchestrationQueue-DLQ --attribute-names QueueArn --query 'Attributes.QueueArn' --output text)
-awslocal sqs create-queue --queue-name TransformOrchestrationQueue --attributes "{\"RedrivePolicy\":\"{\\\"deadLetterTargetArn\\\":\\\"$TRANSFORM_DLQ_ARN\\\",\\\"maxReceiveCount\\\":\\\"3\\\"}\"}"
+    awslocal sqs create-queue --queue-name TransformOrchestrationQueue --attributes "{\"RedrivePolicy\":\"{\\\"deadLetterTargetArn\\\":\\\"$TRANSFORM_DLQ_ARN\\\",\\\"maxReceiveCount\\\":\\\"3\\\"}\"}"
+
+    awslocal sqs create-queue --queue-name TransformComputeQueue-DLQ
+    COMPUTE_DLQ_ARN=$(awslocal sqs get-queue-attributes --queue-url http://localhost:4566/000000000000/TransformComputeQueue-DLQ --attribute-names QueueArn --query 'Attributes.QueueArn' --output text)
+    awslocal sqs create-queue --queue-name TransformComputeQueue --attributes "{\"RedrivePolicy\":\"{\\\"deadLetterTargetArn\\\":\\\"$COMPUTE_DLQ_ARN\\\",\\\"maxReceiveCount\\\":\\\"3\\\"}\"}"
 
 awslocal sqs create-queue --queue-name DeliverQueue-DLQ
 DELIVER_DLQ_ARN=$(awslocal sqs get-queue-attributes --queue-url http://localhost:4566/000000000000/DeliverQueue-DLQ --attribute-names QueueArn --query 'Attributes.QueueArn' --output text)

frontend/web/package.json

@@ -32,6 +32,7 @@
     "date-fns": "^4.4.0",
     "lucide-react": "^1.21.0",
     "next-themes": "^0.4.6",
+    "node-forge": "^1.4.0",
     "oidc-client-ts": "^3.5.0",
     "react": "^19.2.7",
     "react-dom": "^19.2.7",
@@ -44,6 +45,7 @@
   "devDependencies": {
     "@tanstack/router-plugin": "^1.168.19",
     "@types/node": "^24.13.2",
+    "@types/node-forge": "^1.3.14",
     "@types/react": "^19.2.17",
     "@types/react-dom": "^19.2.3",
     "@vitejs/plugin-react": "^6.0.2",

frontend/web/pnpm-lock.yaml

@@ -32,6 +32,7 @@ export interface IPartnersRepository {
   // Certificates
   exportCertificates(partnerId: string): Promise<CertificatesExport>;
   rotateCertificates(partnerId: string, payload: RotateCertPayload): Promise<Partner>;
+  generateCertificate(as2Id: string): Promise<{ public_cert_pem: string; private_key_vault_ref: string }>;
 
   // Tenant Partners
   getTenantPartners(): Promise<Partner[]>;

frontend/web/src/features/partners/api/IPartnersRepository.ts

@@ -228,6 +228,13 @@ export function useRotateCertificatesMutation() {
   );
 }
 
+export function useGenerateCertificateMutation() {
+  const repo = useRepository();
+  return useMutation({
+    mutationFn: (as2Id: string) => repo.generateCertificate(as2Id)
+  });
+}
+
 export function useTestSftpConnectionMutation() {
   const repo = useRepository();
   return useMutation({

frontend/web/src/features/partners/api/partnerHooks.ts

@@ -124,6 +124,13 @@ class HttpPartnersRepository implements IPartnersRepository {
     );
   }
 
+  generateCertificate(as2Id: string): Promise<{ public_cert_pem: string; private_key_vault_ref: string }> {
+    return this.request(
+      `/api/v1/platform/trading-partners/as2/certificates/generate`,
+      { method: 'POST', body: JSON.stringify({ as2_id: as2Id }) },
+    );
+  }
+
   // ── Tenant Partners ────────────────────────
   async getTenantPartners(): Promise<Partner[]> {
     const data = await this.request<any[]>('/api/v1/trading-partners');

frontend/web/src/features/partners/api/partnersApi.ts

@@ -1,20 +1,21 @@
-import { useState } from 'react';
+import React, { useState } from 'react';
 import { useForm, Controller } from 'react-hook-form';
 import { Dialog, DialogContent, DialogHeader, DialogTitle, DialogFooter } from '@/components/ui/dialog';
 import type { AS2Partner } from '../types';
 import { useCertificatesExportQuery, useUpdatePlatformPartnerMutation, useRotateCertificatesMutation } from '../api/partnerHooks';
+import { pki } from 'node-forge';
 import { Copy, Download, Loader2, ChevronDown, ChevronRight, CheckCircle2, Clock, ClipboardPaste } from 'lucide-react';
 import { Button } from '@/components/ui/button';
 import { Input } from '@/components/ui/input';
 import { Label } from '@/components/ui/label';
 import { useToast } from '@/hooks/use-toast';
-import { usePlatformConfig } from '@/features/platform/api/configHooks';
+import { usePlatformSettings } from '@/features/platform/api/settingsHooks';
 import { Combobox } from '@/components/ui/combobox';
 import { CertificateInput } from './CertificateInput';
 
 export function As2PartnerDetails({ partner, onCancel }: { partner: AS2Partner, onCancel?: () => void }) {
   const { toast } = useToast();
-  const { data: platformConfig } = usePlatformConfig();
+  const { data: platformSettings } = usePlatformSettings();
 
   const updatePlatform = useUpdatePlatformPartnerMutation();
   const rotateCertificates = useRotateCertificatesMutation();
@@ -161,7 +162,7 @@ export function As2PartnerDetails({ partner, onCancel }: { partner: AS2Partner,
                 control={control}
                 render={({ field }) => (
                   <Combobox
-                    options={platformConfig?.available_as2_receive_urls || []}
+                    options={platformSettings?.available_as2_receive_urls || []}
                     value={field.value}
                     onChange={field.onChange}
                     placeholder="https://..."
@@ -199,8 +200,11 @@ export function As2PartnerDetails({ partner, onCancel }: { partner: AS2Partner,
               <table className="w-full text-left border-collapse">
                 <thead>
                   <tr className="border-b border-slate-200/60 bg-slate-50/50">
-                    <th className="px-6 py-4 text-xs font-semibold text-slate-500 uppercase tracking-wider">Status</th>
+                    <th className="px-6 py-4 text-xs font-semibold text-slate-500 uppercase tracking-wider w-32">Status</th>
                     <th className="px-6 py-4 text-xs font-semibold text-slate-500 uppercase tracking-wider">Description</th>
+                    <th className="px-6 py-4 text-xs font-semibold text-slate-500 uppercase tracking-wider w-32">Issued</th>
+                    <th className="px-6 py-4 text-xs font-semibold text-slate-500 uppercase tracking-wider w-32">Expires</th>
+                    <th className="px-6 py-4 w-full"></th>
                   </tr>
                 </thead>
                 <tbody className="divide-y divide-slate-100">
@@ -280,6 +284,19 @@ function CertificateRow({
   const [expanded, setExpanded] = useState(false);
   const { toast } = useToast();
 
+  const certInfo = React.useMemo(() => {
+    if (!publicPem || !publicPem.includes('-----BEGIN CERTIFICATE-----')) return null;
+    try {
+      const cert = pki.certificateFromPem(publicPem);
+      return {
+        notBefore: cert.validity.notBefore.toLocaleDateString(undefined, { year: 'numeric', month: 'short', day: 'numeric' }),
+        notAfter: cert.validity.notAfter.toLocaleDateString(undefined, { year: 'numeric', month: 'short', day: 'numeric' }),
+      };
+    } catch (e) {
+      return null;
+    }
+  }, [publicPem]);
+
   const handleCopy = (text: string, label: string) => {
     navigator.clipboard.writeText(text);
     toast({ title: 'Copied', description: `${label} copied to clipboard.` });
@@ -327,21 +344,35 @@ function CertificateRow({
             </span>
           )}
         </td>
-        <td className="px-6 py-4 text-sm text-slate-500 w-full">
-          <div className="flex items-center justify-between">
-            <span>{role === 'Active' ? 'Actively used for signing and decryption' : 'In grace period for legacy traffic'}</span>
-            <div className="flex items-center gap-3">
-              <span className="text-xs font-medium text-slate-400 group-hover:text-indigo-600 transition-colors">
-                {expanded ? 'Hide Details' : 'View Details'}
-              </span>
-              {expanded ? <ChevronDown className="w-4 h-4 text-slate-400 group-hover:text-indigo-600 transition-colors" /> : <ChevronRight className="w-4 h-4 text-slate-400 group-hover:text-indigo-600 transition-colors" />}
-            </div>
+        <td className="px-6 py-4 text-sm text-slate-500 whitespace-nowrap">
+          <span>{role === 'Active' ? 'Actively used for signing and decryption' : 'In grace period for legacy traffic'}</span>
+        </td>
+        <td className="px-6 py-4 whitespace-nowrap">
+          {certInfo ? (
+            <span className="text-sm text-slate-600 font-medium">{certInfo.notBefore}</span>
+          ) : (
+            <span className="text-sm text-slate-400 italic">Unknown</span>
+          )}
+        </td>
+        <td className="px-6 py-4 whitespace-nowrap">
+          {certInfo ? (
+            <span className="text-sm text-slate-600 font-medium">{certInfo.notAfter}</span>
+          ) : (
+            <span className="text-sm text-slate-400 italic">Unknown</span>
+          )}
+        </td>
+        <td className="px-6 py-4 whitespace-nowrap text-right">
+          <div className="flex items-center justify-end gap-3">
+            <span className="text-xs font-medium text-slate-400 group-hover:text-indigo-600 transition-colors">
+              {expanded ? 'Hide Details' : 'View Details'}
+            </span>
+            {expanded ? <ChevronDown className="w-4 h-4 text-slate-400 group-hover:text-indigo-600 transition-colors" /> : <ChevronRight className="w-4 h-4 text-slate-400 group-hover:text-indigo-600 transition-colors" />}
           </div>
         </td>
       </tr>
       {expanded && (
         <tr>
-          <td colSpan={2} className="p-0 border-t-0">
+          <td colSpan={5} className="p-0 border-t-0">
             <div className="bg-slate-50 p-4 border-b border-slate-100 flex flex-col gap-4 shadow-inner">
 
               <div className="flex flex-col gap-2">