translate.ts

import type {
  ActivityEventInput,
  Actor,
  OfferPatch,
  OfferSnapshot,
} from '../events/types';

export const PARTNER_EVENT_NAMES = [
  'attribution_verified',
  'milestone_completed',
  'attribution_failed',
  'reward_paid',
] as const;

export type PartnerEventName = (typeof PARTNER_EVENT_NAMES)[number];

export type PartnerWebhookPayload = {
  partner_slug: string;
  event_name: PartnerEventName;
  offer_id: string;
  event_id?: string;
  idempotency_key: string;
};

export type TranslationContext = {
  snapshot: OfferSnapshot;
};

export type TranslatedEvents = {
  audit: ActivityEventInput;
  effects: ActivityEventInput[];
};

export type TranslationErrorCode = 'MISSING_EVENT_ID' | 'UNKNOWN_EVENT_ID';

export class TranslationError extends Error {
  constructor(
    message: string,
    public readonly code: TranslationErrorCode,
  ) {
    super(message);
    this.name = 'TranslationError';
  }
}

export function translatePartnerWebhook(
  payload: PartnerWebhookPayload,
  context: TranslationContext,
): TranslatedEvents {
  const actor: Actor = { kind: 'partner', slug: payload.partner_slug };
  const { snapshot } = context;

  const audit: ActivityEventInput = {
    kind: 'partner_webhook_received',
    actor,
    offer_id: payload.offer_id,
    event_name: payload.event_name,
    idempotency_key: payload.idempotency_key,
  };

  const effects: ActivityEventInput[] = [];

  // On Vercel Fluid Compute, partner webhooks and the SSE/polling channel
  // the client hydrates from can land on different Node.js instances with
  // divergent in-memory state. The patch below must describe the canonical
  // target state — not a diff against *this* lambda's snapshot — so a
  // client whose view advanced on a different lambda still reconciles to
  // the right bucket. Always emitting _bucket is a no-op when they agree
  // and a self-heal when they don't.
  switch (payload.event_name) {
    case 'attribution_verified': {
      const patch: OfferPatch = {
        user_offer_activity: { status: 'IN_PROGRESS' },
        _bucket: 'in_progress',
      };
      if (snapshot._startedAtMinutesAgo == null) {
        patch._startedAtMinutesAgo = 0;
      }
      effects.push({ kind: 'offer_updated', actor, offer_id: payload.offer_id, patch });
      effects.push({
        kind: 'activity_status_changed',
        actor,
        offer_id: payload.offer_id,
        from: snapshot.user_offer_activity.status,
        to: 'IN_PROGRESS',
        reason: 'attribution_verified',
      });
      break;
    }
    case 'attribution_failed': {
      const patch: OfferPatch = {
        user_offer_activity: { status: 'RETRY' },
        _bucket: 'pending',
      };
      effects.push({ kind: 'offer_updated', actor, offer_id: payload.offer_id, patch });
      effects.push({
        kind: 'activity_status_changed',
        actor,
        offer_id: payload.offer_id,
        from: snapshot.user_offer_activity.status,
        to: 'RETRY',
        reason: 'attribution_failed',
      });
      break;
    }
    case 'reward_paid': {
      const patch: OfferPatch = {
        user_offer_activity: { status: 'COMPLETED' },
        _bucket: 'completed',
      };
      if (snapshot._lastCompletedAtMinutesAgo == null) {
        patch._lastCompletedAtMinutesAgo = 0;
      }
      effects.push({ kind: 'offer_updated', actor, offer_id: payload.offer_id, patch });
      effects.push({
        kind: 'activity_status_changed',
        actor,
        offer_id: payload.offer_id,
        from: snapshot.user_offer_activity.status,
        to: 'COMPLETED',
        reason: 'reward_paid',
      });
      break;
    }
    case 'milestone_completed': {
      if (!payload.event_id) {
        throw new TranslationError(
          'milestone_completed requires event_id',
          'MISSING_EVENT_ID',
        );
      }
      const matching = snapshot.events.find((e) => e.id === payload.event_id);
      if (!matching) {
        throw new TranslationError(
          `Unknown event_id "${payload.event_id}" for offer "${payload.offer_id}"`,
          'UNKNOWN_EVENT_ID',
        );
      }
      const updatedEvents = snapshot.events.map((e) =>
        e.id === payload.event_id ? { ...e, is_completed: true } : e,
      );
      effects.push({
        kind: 'offer_updated',
        actor,
        offer_id: payload.offer_id,
        patch: { events: updatedEvents },
      });
      effects.push({
        kind: 'event_milestone_completed',
        actor,
        offer_id: payload.offer_id,
        event_id: payload.event_id,
        payout: matching.payout,
      });
      break;
    }
  }

  return { audit, effects };
}