Feat: analytics-webhoo-event-sourcing-groups/family: Subtrackr (#469)

Author: Dannyswiss1

backend/ml/churnModel.py

@@ -85,6 +85,35 @@ def _get_recommended_action(self, risk_level: str, top_factors: List[Dict]) -> s
         else:
             return "Offer a 1-month free subscription to retain user."
 
+
+class RevenueForecastModel:
+    def forecast(self, observations: List[Dict], horizon: int = 3) -> List[Dict]:
+        values = [float(item.get("revenue", 0)) for item in observations]
+        if not values:
+            return []
+
+        latest = values[-1]
+        deltas = [values[index] - values[index - 1] for index in range(1, len(values))]
+        average_delta = sum(deltas) / len(deltas) if deltas else 0
+        variance = (
+            sum((delta - average_delta) ** 2 for delta in deltas) / len(deltas)
+            if deltas
+            else max(latest * 0.05, 1)
+        )
+        deviation = math.sqrt(variance)
+
+        forecast = []
+        for step in range(1, horizon + 1):
+            expected = max(0, latest + average_delta * step)
+            confidence = deviation * math.sqrt(step) * 1.96
+            forecast.append({
+                "period": f"forecast_{step}",
+                "expected_revenue": round(expected, 2),
+                "lower_bound": round(max(0, expected - confidence), 2),
+                "upper_bound": round(expected + confidence, 2),
+            })
+        return forecast
+
 if __name__ == "__main__":
     model = ChurnPredictionModel()
     test_data = {

backend/services/index.ts

@@ -19,3 +19,13 @@ export {
   isWebhookEventAllowed,
 } from './webhook';
 export type { RegisterWebhookInput, WebhookDeliveryResult, WebhookEventInput } from './webhook';
+export {
+  SubscriptionEventStore,
+  subscriptionEventStore,
+} from './subscriptionEventStore';
+export type {
+  SubscriptionEvent,
+  SubscriptionEventPage,
+  SubscriptionEventQuery,
+  SubscriptionEventType,
+} from './subscriptionEventStore';

backend/services/predictionService.ts

@@ -21,6 +21,18 @@ export interface UserChurnData {
   priceSensitivityIndex: number;
 }
 
+export interface RevenueObservation {
+  period: string;
+  revenue: number;
+}
+
+export interface ForecastPoint {
+  period: string;
+  expectedRevenue: number;
+  lowerBound: number;
+  upperBound: number;
+}
+
 export class PredictionService {
   // Path for future Python bridge integration
   private static readonly _PYTHON_PATH = path.join(__dirname, '../ml/churnModel.py');
@@ -81,4 +93,34 @@ export class PredictionService {
       throw new Error('Failed to fetch risk factors');
     }
   }
+
+  static async forecastRevenue(
+    observations: RevenueObservation[],
+    horizon = 3
+  ): Promise<ForecastPoint[]> {
+    if (observations.length === 0) return [];
+
+    const values = observations.map((entry) => entry.revenue);
+    const latest = values[values.length - 1];
+    const deltas = values.slice(1).map((value, index) => value - values[index]);
+    const averageDelta = deltas.length
+      ? deltas.reduce((sum, delta) => sum + delta, 0) / deltas.length
+      : 0;
+    const variance = deltas.length
+      ? deltas.reduce((sum, delta) => sum + Math.pow(delta - averageDelta, 2), 0) / deltas.length
+      : Math.max(latest * 0.05, 1);
+    const deviation = Math.sqrt(variance);
+
+    return Array.from({ length: horizon }, (_, index) => {
+      const step = index + 1;
+      const expectedRevenue = Math.max(0, latest + averageDelta * step);
+      const confidence = deviation * Math.sqrt(step) * 1.96;
+      return {
+        period: `forecast_${step}`,
+        expectedRevenue: Number(expectedRevenue.toFixed(2)),
+        lowerBound: Number(Math.max(0, expectedRevenue - confidence).toFixed(2)),
+        upperBound: Number((expectedRevenue + confidence).toFixed(2)),
+      };
+    });
+  }
 }

backend/services/subscriptionEventStore.ts

@@ -0,0 +1,108 @@
+export type SubscriptionEventType =
+  | 'subscription.created'
+  | 'subscription.updated'
+  | 'subscription.renewed'
+  | 'subscription.cancelled'
+  | 'subscription.payment_failed'
+  | 'subscription.upgraded'
+  | 'subscription.paused'
+  | 'subscription.resumed';
+
+export interface SubscriptionEvent<TPayload extends Record<string, unknown> = Record<string, unknown>> {
+  id: string;
+  subscriptionId: string;
+  sequence: number;
+  type: SubscriptionEventType;
+  payload: TPayload;
+  occurredAt: number;
+  schemaVersion: number;
+  archivedAt?: number;
+}
+
+export interface SubscriptionEventQuery {
+  subscriptionId?: string;
+  type?: SubscriptionEventType;
+  from?: number;
+  to?: number;
+  limit?: number;
+  cursor?: number;
+  includeArchived?: boolean;
+}
+
+export interface SubscriptionEventPage {
+  events: SubscriptionEvent[];
+  nextCursor?: number;
+}
+
+export class SubscriptionEventStore {
+  private readonly events: SubscriptionEvent[] = [];
+  private readonly sequenceBySubscription = new Map<string, number>();
+
+  append<TPayload extends Record<string, unknown> = Record<string, unknown>>(
+    event: Omit<SubscriptionEvent<TPayload>, 'id' | 'sequence' | 'occurredAt' | 'schemaVersion'> &
+      Partial<Pick<SubscriptionEvent, 'occurredAt' | 'schemaVersion'>>
+  ): SubscriptionEvent<TPayload> {
+    const nextSequence = (this.sequenceBySubscription.get(event.subscriptionId) ?? 0) + 1;
+    this.sequenceBySubscription.set(event.subscriptionId, nextSequence);
+
+    const record: SubscriptionEvent<TPayload> = {
+      ...event,
+      id: `sev_${Date.now().toString(36)}_${nextSequence}`,
+      sequence: nextSequence,
+      occurredAt: event.occurredAt ?? Date.now(),
+      schemaVersion: event.schemaVersion ?? 1,
+    };
+    this.events.push(record);
+    return record;
+  }
+
+  query(query: SubscriptionEventQuery = {}): SubscriptionEventPage {
+    const cursor = query.cursor ?? 0;
+    const limit = Math.max(1, query.limit ?? 50);
+    const filtered = this.events.filter((event) => {
+      if (!query.includeArchived && event.archivedAt) return false;
+      if (query.subscriptionId && event.subscriptionId !== query.subscriptionId) return false;
+      if (query.type && event.type !== query.type) return false;
+      if (query.from && event.occurredAt < query.from) return false;
+      if (query.to && event.occurredAt > query.to) return false;
+      return true;
+    });
+    const events = filtered.slice(cursor, cursor + limit);
+    const nextCursor = cursor + limit < filtered.length ? cursor + limit : undefined;
+    return { events, nextCursor };
+  }
+
+  reconstruct(subscriptionId: string): Record<string, unknown> {
+    return this.query({ subscriptionId, includeArchived: true, limit: Number.MAX_SAFE_INTEGER })
+      .events.sort((a, b) => a.sequence - b.sequence)
+      .reduce<Record<string, unknown>>(
+        (state, event) => ({
+          ...state,
+          ...(event.payload as Record<string, unknown>),
+          id: subscriptionId,
+          lastEventType: event.type,
+          updatedAt: event.occurredAt,
+        }),
+        { id: subscriptionId }
+      );
+  }
+
+  replay(subscriptionId: string, handler: (event: SubscriptionEvent) => void): void {
+    this.query({ subscriptionId, includeArchived: true, limit: Number.MAX_SAFE_INTEGER })
+      .events.sort((a, b) => a.sequence - b.sequence)
+      .forEach(handler);
+  }
+
+  archiveBefore(timestamp: number): number {
+    let archived = 0;
+    for (const event of this.events) {
+      if (!event.archivedAt && event.occurredAt < timestamp) {
+        event.archivedAt = Date.now();
+        archived += 1;
+      }
+    }
+    return archived;
+  }
+}
+
+export const subscriptionEventStore = new SubscriptionEventStore();

backend/services/webhook.ts

@@ -20,6 +20,7 @@ export interface RegisterWebhookInput {
   events: WebhookEventType[];
   secretKey: string;
   retryPolicy?: Partial<WebhookRetryPolicy>;
+  rateLimitPerMinute?: number;
   isPaused?: boolean;
 }
 
@@ -94,6 +95,7 @@ export class WebhookDeliveryService {
   private readonly webhooks = new Map<string, WebhookConfig>();
   private readonly deliveries = new Map<string, WebhookDelivery>();
   private readonly deliveredKeys = new Set<string>();
+  private readonly rateLimitWindows = new Map<string, number[]>();
 
   constructor(options: { fetchImpl?: FetchLike; sleepImpl?: (ms: number) => Promise<void> } = {}) {
     this.fetchImpl = options.fetchImpl ?? fetch;
@@ -110,6 +112,7 @@ export class WebhookDeliveryService {
       events: [...input.events],
       secretKey: input.secretKey,
       retryPolicy: clampRetryPolicy(input.retryPolicy),
+      rateLimitPerMinute: input.rateLimitPerMinute,
       isPaused: input.isPaused ?? false,
       createdAt,
       updatedAt: createdAt,
@@ -136,6 +139,7 @@ export class WebhookDeliveryService {
       events: input.events ? [...input.events] : existing.events,
       secretKey: input.secretKey ?? existing.secretKey,
       retryPolicy: clampRetryPolicy(input.retryPolicy ?? existing.retryPolicy),
+      rateLimitPerMinute: input.rateLimitPerMinute ?? existing.rateLimitPerMinute,
       isPaused: input.isPaused ?? existing.isPaused,
       updatedAt: now(),
     };
@@ -193,6 +197,9 @@ export class WebhookDeliveryService {
     const avgAttempts = totalDeliveries
       ? deliveries.reduce((sum, d) => sum + d.attempts, 0) / totalDeliveries
       : 0;
+    const latencySamples = deliveries
+      .map((delivery) => delivery.latencyMs)
+      .filter((latency): latency is number => typeof latency === 'number');
 
     return {
       webhookId,
@@ -203,6 +210,9 @@ export class WebhookDeliveryService {
       pendingDeliveries,
       successRate: totalDeliveries ? successfulDeliveries / totalDeliveries : 0,
       avgAttempts,
+      avgLatencyMs: latencySamples.length
+        ? latencySamples.reduce((sum, latency) => sum + latency, 0) / latencySamples.length
+        : 0,
       lastSuccessAt: deliveries
         .filter((delivery) => delivery.status === 'delivered' && delivery.deliveredAt)
         .map((delivery) => delivery.deliveredAt as number)
@@ -270,6 +280,28 @@ export class WebhookDeliveryService {
       return { delivery };
     }
 
+    if (this.isRateLimited(webhook)) {
+      const delivery: WebhookDelivery = {
+        id: createId('del'),
+        webhookId: webhook.id,
+        eventId: payload.id,
+        eventType: payload.eventType,
+        url: webhook.url,
+        payload,
+        status: 'retrying',
+        attempts: 0,
+        maxAttempts: webhook.retryPolicy.maxRetries,
+        createdAt: now(),
+        updatedAt: now(),
+        signature,
+        idempotencyKey,
+        errorMessage: 'Webhook endpoint rate limited',
+        nextRetryAt: now() + 60_000,
+      };
+      this.deliveries.set(delivery.id, delivery);
+      return { delivery };
+    }
+
     const delivery: WebhookDelivery = {
       id: createId('del'),
       webhookId: webhook.id,
@@ -375,6 +407,7 @@ export class WebhookDeliveryService {
             status: 'delivered',
             responseCode: response.status,
             deliveredAt: now(),
+            latencyMs: now() - attemptAt,
           },
           response
         );
@@ -442,6 +475,21 @@ export class WebhookDeliveryService {
     const rawDelay = Math.floor(policy.initialDelayMs * Math.pow(factor, Math.max(0, attempt - 1)));
     return Math.min(rawDelay, policy.maxDelayMs);
   }
+
+  private isRateLimited(webhook: WebhookConfig): boolean {
+    if (!webhook.rateLimitPerMinute || webhook.rateLimitPerMinute <= 0) return false;
+    const windowStart = now() - 60_000;
+    const current = (this.rateLimitWindows.get(webhook.id) ?? []).filter(
+      (timestamp) => timestamp >= windowStart
+    );
+    if (current.length >= webhook.rateLimitPerMinute) {
+      this.rateLimitWindows.set(webhook.id, current);
+      return true;
+    }
+    current.push(now());
+    this.rateLimitWindows.set(webhook.id, current);
+    return false;
+  }
 }
 
 export const webhookDeliveryService = new WebhookDeliveryService();

contracts/subscription/src/events.rs

@@ -1,9 +1,68 @@
-use soroban_sdk::{Address, Env};
+use soroban_sdk::{contracttype, Address, Env, String, Vec};
 use subtrackr_types::{
     Plan, Subscription, SubscriptionStatus, WebhookEventPayload, WebhookEventType,
     WebhookPlanSnapshot, WebhookSubscriptionSnapshot,
 };
 
+#[contracttype]
+#[derive(Clone, Debug, PartialEq)]
+pub enum SubscriptionEventType {
+    Created,
+    Updated,
+    Renewed,
+    Cancelled,
+    PaymentFailed,
+    Upgraded,
+    Paused,
+    Resumed,
+    RefundRequested,
+    RefundApproved,
+    RefundRejected,
+    TransferRequested,
+    TransferAccepted,
+}
+
+#[contracttype]
+#[derive(Clone, Debug, PartialEq)]
+pub struct SubscriptionAuditEvent {
+    pub id: u64,
+    pub subscription_id: u64,
+    pub sequence: u64,
+    pub event_type: SubscriptionEventType,
+    pub actor: Address,
+    pub occurred_at: u64,
+    pub schema_version: u32,
+    pub payload_hash: String,
+}
+
+pub(crate) fn build_audit_event(
+    env: &Env,
+    subscription_id: u64,
+    sequence: u64,
+    event_type: SubscriptionEventType,
+    actor: &Address,
+    payload_hash: String,
+) -> SubscriptionAuditEvent {
+    SubscriptionAuditEvent {
+        id: env.ledger().sequence() as u64,
+        subscription_id,
+        sequence,
+        event_type,
+        actor: actor.clone(),
+        occurred_at: env.ledger().timestamp(),
+        schema_version: 1,
+        payload_hash,
+    }
+}
+
+pub(crate) fn replay_state(events: Vec<SubscriptionAuditEvent>) -> Option<SubscriptionEventType> {
+    let mut latest: Option<SubscriptionEventType> = None;
+    for event in events.iter() {
+        latest = Some(event.event_type);
+    }
+    latest
+}
+
 pub(crate) fn subscription_snapshot(sub: &Subscription) -> WebhookSubscriptionSnapshot {
     WebhookSubscriptionSnapshot {
         id: sub.id,