Orion Notification System — Execution Plan

01Task summary

02NS-01 — Create Pub/Sub topics and service account Low risk

Goal

Provision the three GCP Pub/Sub topics in the metiss-dev project, create a dedicated service account for push authentication, and register placeholder push subscriptions. No code changes — infra only.

Commands

# Topics
gcloud pubsub topics create platform.jobs     --project=metiss-dev
gcloud pubsub topics create platform.invoices --project=metiss-dev
gcloud pubsub topics create platform.auth     --project=metiss-dev

# Service account for push auth
gcloud iam service-accounts create pubsub-push-sa \
  --display-name="Pub/Sub Push Invoker" \
  --project=metiss-dev

# Grant invoker role on the notification-service Cloud Run service
# (run after NS-02 deploys the service)
gcloud run services add-iam-policy-binding notification-service \
  --region=us-central1 \
  --member="serviceAccount:pubsub-push-sa@metiss-dev.iam.gserviceaccount.com" \
  --role="roles/run.invoker"

# Push subscriptions (one per topic)
# Replace <CLOUD_RUN_URL> with the actual URL after NS-02 deploys
for TOPIC in platform.jobs platform.invoices platform.auth; do
  gcloud pubsub subscriptions create "notif-${TOPIC//./-}-push" \
    --topic="$TOPIC" \
    --push-endpoint="<CLOUD_RUN_URL>/events/push" \
    --push-auth-service-account="pubsub-push-sa@metiss-dev.iam.gserviceaccount.com" \
    --ack-deadline=60 \
    --min-retry-delay=10s \
    --max-retry-delay=600s \
    --project=metiss-dev
done

Acceptance criteria

• Three topics visible in GCP Console → Pub/Sub → Topics
• Service account pubsub-push-sa@metiss-dev.iam.gserviceaccount.com exists
• Three push subscriptions created (endpoint URL can be updated after NS-02)

03NS-02 — Scaffold notification-service Low risk

Goal

Create the notification-service directory in the Metiss-AI/api monorepo. The service should deploy to Cloud Run and return HTTP 200 on GET /health — nothing more. CI/CD, Dockerfile, and Secret Manager wiring are included in this task so all subsequent tasks can deploy incrementally.

Directory structure

api/notification-service/
├── Dockerfile
├── requirements.txt
├── main.py                  # FastAPI app entrypoint
├── alembic.ini
├── alembic/
│   ├── env.py
│   └── versions/            # migration files added in NS-03
├── app/
│   ├── config.py            # pydantic-settings: DB_URL, RESEND_API_KEY, etc.
│   ├── database.py          # asyncpg connection pool
│   ├── routers/
│   │   ├── health.py        # GET /health
│   │   ├── push.py          # POST /events/push  (NS-04)
│   │   ├── webhook.py       # POST /events/webhook  (NS-09)
│   │   ├── preferences.py   # GET/PUT /preferences/{user_id}  (NS-08)
│   │   └── unsubscribe.py   # GET /unsubscribe  (NS-08)
│   └── services/
│       ├── email.py         # Resend client wrapper  (NS-05)
│       ├── idempotency.py   # notification_log dedup  (NS-04)
│       ├── preferences.py   # user_notification_preference queries  (NS-08)
│       └── templates.py     # Jinja2 loader  (NS-05)
└── templates/               # Jinja2 HTML + txt files  (NS-06, NS-07)

Key files

# requirements.txt (minimum for scaffold)
fastapi==0.115.0
uvicorn[standard]==0.30.0
pydantic-settings==2.3.0
asyncpg==0.29.0
alembic==1.13.0
google-auth==2.32.0        # OIDC token verification (NS-04)
resend==2.3.0              # email provider (NS-05)
jinja2==3.1.4              # templates (NS-05)

# main.py
from fastapi import FastAPI
from app.routers import health
from app.database import init_pool, close_pool

app = FastAPI()
app.include_router(health.router)

@app.on_event("startup")
async def startup():
    await init_pool()

@app.on_event("shutdown")
async def shutdown():
    await close_pool()

Secret Manager keys to provision

Acceptance criteria

• GET /health returns { "status": "ok" } on the deployed Cloud Run URL
• CI/CD pipeline deploys on every push to main in the api repo
• All four Secret Manager secrets exist in metiss-dev
• Update NS-01 push subscription endpoint URL to the new Cloud Run URL

04NS-03 — Database migrations: notification tables Low risk

Goal

Create three tables in the shared Cloud SQL instance via Alembic. These tables are owned by the notification service; the Orion BFF does not query them directly.

Migration SQL

-- notification_log (append-only delivery record)
CREATE TABLE notification_log (
  id                  UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  idempotency_key     TEXT NOT NULL UNIQUE,
  event_type          TEXT NOT NULL,
  event_id            TEXT NOT NULL,
  recipient_user_id   UUID NOT NULL,
  recipient_email     TEXT NOT NULL,
  channel             TEXT NOT NULL DEFAULT 'email',
  status              TEXT NOT NULL,
  provider_email_id   TEXT,
  sent_at             TIMESTAMPTZ,
  delivered_at        TIMESTAMPTZ,
  opened_at           TIMESTAMPTZ,
  error_message       TEXT,
  created_at          TIMESTAMPTZ NOT NULL DEFAULT now()
);

CREATE INDEX notification_log_idempotency_key_idx
  ON notification_log (idempotency_key);

CREATE INDEX notification_log_provider_email_id_idx
  ON notification_log (provider_email_id)
  WHERE provider_email_id IS NOT NULL;

-- user_notification_preference (opt-out store; row absent = opted in)
CREATE TABLE user_notification_preference (
  id            UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  user_id       UUID NOT NULL,
  event_type    TEXT NOT NULL,
  email_enabled BOOLEAN NOT NULL DEFAULT TRUE,
  used_nonce    TEXT,
  updated_at    TIMESTAMPTZ NOT NULL DEFAULT now(),
  UNIQUE (user_id, event_type)
);

-- notification_template (v2 placeholder — not queried in v1)
CREATE TABLE notification_template (
  id          UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  event_type  TEXT NOT NULL UNIQUE,
  subject     TEXT NOT NULL,
  html_body   TEXT NOT NULL,
  text_body   TEXT NOT NULL,
  active      BOOLEAN NOT NULL DEFAULT FALSE,
  updated_at  TIMESTAMPTZ NOT NULL DEFAULT now()
);

Acceptance criteria

• All three tables exist in Cloud SQL
• notification_log_idempotency_key_idx unique index exists
• Alembic migration version is committed and alembic upgrade head is idempotent

05NS-04 — Implement POST /events/push Medium risk

Goal

The core of the notification service. This endpoint receives all Pub/Sub push messages, verifies their authenticity, decodes the envelope, checks for duplicates, and hands off to the appropriate event handler. Event handlers are stubs in this task — Resend integration is wired in NS-05.

Implementation

# app/routers/push.py

from fastapi import APIRouter, Request, HTTPException
from google.auth.transport import requests as google_requests
from google.oauth2 import id_token
import base64, json
from app.services.idempotency import is_duplicate, record_attempt
from app.config import settings

router = APIRouter()

AUDIENCE = settings.cloud_run_url  # e.g. https://notification-service-xxx.run.app

@router.post("/events/push")
async def handle_push(request: Request):
    # 1. Verify OIDC token
    auth = request.headers.get("Authorization", "")
    if not auth.startswith("Bearer "):
        raise HTTPException(status_code=401, detail="Missing bearer token")
    try:
        id_token.verify_oauth2_token(
            auth.removeprefix("Bearer "),
            google_requests.Request(),
            AUDIENCE,
        )
    except Exception:
        raise HTTPException(status_code=401, detail="Invalid OIDC token")

    # 2. Decode Pub/Sub envelope
    body = await request.json()
    message = body.get("message", {})
    message_id = message.get("messageId")
    data = json.loads(base64.b64decode(message.get("data", "")))

    # 3. Idempotency check
    if await is_duplicate(message_id):
        return {"status": "duplicate"}

    # 4. Route to handler
    event_type = data.get("event_type", "")
    try:
        await dispatch(event_type, data)
    except PermanentError as e:
        # Acknowledge to prevent infinite retry; log the failure
        await record_attempt(message_id, data, status="failed", error=str(e))
        return {"status": "failed", "reason": str(e)}

    return {"status": "ok"}

Idempotency service

# app/services/idempotency.py

async def is_duplicate(message_id: str) -> bool:
    row = await db.fetchrow(
        "SELECT id FROM notification_log WHERE idempotency_key = $1",
        message_id,
    )
    return row is not None

Acceptance criteria

• Request with valid OIDC token + well-formed envelope returns 200
• Request with missing/invalid OIDC token returns 401
• Sending the same messageId twice returns 200 with { "status": "duplicate" } on the second call
• Malformed base64 data returns 200 with { "status": "failed" } (not 500)

06NS-05 — Resend integration + Jinja2 template system Low risk

Goal

Wire the Resend Python SDK and Jinja2 template loader. After this task, calling send_email(event_type, recipient_email, context) renders the template and dispatches via Resend. No templates are shipped yet — those are NS-06 and NS-07.

Implementation

# app/services/email.py
import resend
from jinja2 import Environment, FileSystemLoader
from app.config import settings

resend.api_key = settings.resend_api_key
SENDER = "Metiss AI <notifications@metiss.ai>"

_env = Environment(loader=FileSystemLoader("templates"), autoescape=True)

async def send_email(
    event_type: str,
    recipient_email: str,
    context: dict,
    *,
    idempotency_key: str,
) -> str:
    slug = event_type.replace(".", "_")
    html = _env.get_template(f"{slug}.html").render(**context)
    text = _env.get_template(f"{slug}.txt").render(**context)
    subject = _env.get_template(f"{slug}.subject.txt").render(**context)

    params = {
        "from": SENDER,
        "to": [recipient_email],
        "subject": subject,
        "html": html,
        "text": text,
        "headers": {
            "List-Unsubscribe": build_unsubscribe_header(
                context["user_id"], event_type
            ),
        },
        "idempotencyKey": idempotency_key,  # Resend dedup
    }
    response = resend.Emails.send(params)
    return response["id"]  # provider_email_id

Template naming convention

Each event type requires three files in notification-service/templates/:

Acceptance criteria

• send_email("job.assigned", "test@example.com", ...) calls Resend and returns a non-empty email_id (use Resend test mode / sandbox domain)
• Missing template file raises a descriptive error (not a 500 to the push endpoint)
• Resend idempotencyKey is set on every call

07NS-06 — Job and invoice email templates Low risk

Goal

Create Jinja2 templates and handler functions for all job and invoice events. After this task, the notification service can receive and send emails for every event in the job and invoice domains. The Orion BFF does not publish these events yet — that is NS-12 and NS-13.

Templates to create

Handler pattern

Each event handler fetches the data it needs from Cloud SQL, builds the context dict, resolves recipients, checks preferences, and calls send_email. Add one handler per event type to app/handlers/:

# app/handlers/job_assigned.py

async def handle(payload: dict, message_id: str):
    job = await db.fetchrow(
        'SELECT * FROM "Job" WHERE "jobId" = $1', payload["job_id"]
    )
    technician = await db.fetchrow(
        'SELECT * FROM "User" WHERE id = $1', payload["recipient_ids"][0]
    )
    if not await preference_allows(technician["id"], "job.assigned"):
        return

    context = {
        "user_id": str(technician["id"]),
        "technician_name": technician["name"],
        "job_id": job["jobId"],
        "job_address": job["customerAddress"],
        "orion_url": "https://orion.metiss.ai",
    }
    email_id = await send_email(
        "job.assigned", technician["email"], context,
        idempotency_key=message_id,
    )
    await record_success(message_id, payload, technician["id"],
                         technician["email"], email_id)

Acceptance criteria

• All 7 templates exist (subject, html, txt = 21 files total)
• Posting a synthetic job.assigned push message to the running service results in an email delivered to the test recipient
• A recipient who has opted out of job.assigned in user_notification_preference does not receive the email

08NS-07 — Auth email templates (replace ad-hoc SMTP) Low risk

Goal

Create templates and handlers for the three auth events. These replace the ad-hoc SMTP calls currently in the Orion API service. The notification service side is built here (NS-07); the Orion side is wired in NS-11.

Templates to create

Acceptance criteria

• All 3 auth templates exist (9 files)
• Posting a synthetic auth.invite_sent message results in an invite email to the test address
• Auth emails do not include an unsubscribe link or header
• Posting an auth.invite_sent event for a user with email notifications disabled still sends the email

09NS-08 — Implement GET/PUT /preferences and GET /unsubscribe Low risk

Goal

Two self-contained endpoints: the preferences API (called by the Orion settings UI) and the one-click unsubscribe endpoint (called from email footer links).

Preferences API

# GET /preferences/{user_id}
# Returns all preference rows for the user (absent rows = opted in by default)

# PUT /preferences/{user_id}
# Body: { "event_type": "job.assigned", "email_enabled": false }
# Upserts into user_notification_preference
# Auth: internal service token in Authorization header (not public)

Unsubscribe endpoint

# GET /unsubscribe?user_id=<uuid>&event_type=<type>&nonce=<ulid>&sig=<hex>

import hmac, hashlib
from app.config import settings

def verify_sig(user_id: str, event_type: str, nonce: str, sig: str) -> bool:
    expected = hmac.new(
        settings.unsubscribe_hmac_secret.encode(),
        f"{user_id}:{event_type}:{nonce}".encode(),
        hashlib.sha256,
    ).hexdigest()
    return hmac.compare_digest(expected, sig)

# On valid sig:
# 1. Check nonce not already used (used_nonce column in user_notification_preference)
# 2. Upsert user_notification_preference with email_enabled=False, used_nonce=nonce
# 3. Return a simple HTML confirmation page

Build the unsubscribe URL at send time

# In app/services/email.py — add alongside send_email()

def build_unsubscribe_url(user_id: str, event_type: str) -> str:
    nonce = str(ulid.new())
    sig = hmac.new(
        settings.unsubscribe_hmac_secret.encode(),
        f"{user_id}:{event_type}:{nonce}".encode(),
        hashlib.sha256,
    ).hexdigest()
    return (
        f"https://notification-service-xxx.run.app/unsubscribe"
        f"?user_id={user_id}&event_type={event_type}&nonce={nonce}&sig={sig}"
    )

Acceptance criteria

• PUT /preferences/{user_id} with a valid internal token creates/updates the preference row
• GET /unsubscribe with a valid HMAC sets email_enabled = false for the given user and event type
• GET /unsubscribe with a tampered sig returns 403
• Replaying the same unsubscribe link (same nonce) after it has been used returns 400

10NS-09 — Implement POST /events/webhook Low risk

Goal

Receive Resend delivery webhooks (email.delivered, email.bounced, email.opened, email.complained) and update notification_log. This closes the delivery tracking loop.

Implementation

# app/routers/webhook.py
from svix.webhooks import Webhook
from app.config import settings

wh = Webhook(settings.resend_webhook_secret)

@router.post("/events/webhook")
async def handle_webhook(request: Request):
    payload = await request.body()
    headers = dict(request.headers)

    try:
        event = wh.verify(payload, headers)  # raises on invalid sig
    except Exception:
        raise HTTPException(status_code=403, detail="Invalid webhook signature")

    email_id   = event["data"]["email_id"]
    event_type = event["type"]  # e.g. "email.delivered"

    status_map = {
        "email.delivered": "delivered",
        "email.bounced":   "bounced",
        "email.complained": "complained",
        "email.opened":    None,  # sets opened_at only
    }
    if event_type not in status_map:
        return {"status": "ignored"}

    if event_type == "email.opened":
        await db.execute(
            "UPDATE notification_log SET opened_at = now() WHERE provider_email_id = $1",
            email_id,
        )
    else:
        await db.execute(
            """UPDATE notification_log
               SET status = $1,
                   delivered_at = CASE WHEN $1 = 'delivered' THEN now() END
               WHERE provider_email_id = $2""",
            status_map[event_type], email_id,
        )
    return {"status": "ok"}

Acceptance criteria

• Sending a Svix-signed test webhook for email.delivered updates the matching notification_log row to status = delivered
• An unsigned or tampered webhook returns 403
• Unknown event types return 200 with { "status": "ignored" }

11NS-10 — Switch Orion Pub/Sub driver to gcp-pubsub Medium risk

Goal

Change the EVENT_PUBLISHER_DRIVER environment variable in the Orion BFF Cloud Run service from local to gcp-pubsub. The abstraction already exists in Orion — this is a configuration change, not a code change. Existing publish calls (if any) will start flowing to Pub/Sub. New publish calls are added in NS-11 through NS-13.

Steps

1. Read backend/api/lib/events/publisher.ts in the orion repo. Confirm thegcp-pubsub driver branch instantiates a Pub/Sub client and calls topic.publishMessage() correctly.
2. Confirm the GCP service account used by the Orion Cloud Run service has the roles/pubsub.publisher IAM role on the three topics created in NS-01.
3. Update the Cloud Run environment variable:

   gcloud run services update orion-bff \
     --region=us-central1 \
     --set-env-vars EVENT_PUBLISHER_DRIVER=gcp-pubsub \
     --project=metiss-dev

4. Verify in GCP Console → Pub/Sub → Topics that no messages are being published yet (no Orion code publishes to these topics before NS-11/12/13). This confirms the driver switch is inert until publish calls are added.

Acceptance criteria

• EVENT_PUBLISHER_DRIVER=gcp-pubsub in the Orion BFF Cloud Run service config
• Orion BFF health check returns 200 after the env var change
• No existing Orion functionality is broken (run smoke tests against staging)
• Orion BFF service account has roles/pubsub.publisher on all three topics

12NS-11 — Publish auth events from Orion BFF Medium risk

Goal

Replace the ad-hoc SMTP calls for user invites and password resets with Pub/Sub publish calls to platform.auth. This is the highest-stakes task because it changes existing production behaviour — test in staging first.

Find and replace the ad-hoc SMTP calls

# Locate all direct email send calls in the Orion BFF
grep -rn "sendMail|nodemailer|smtp|invite.*email|password.*reset.*email" \
  backend/api/ --include="*.ts"

# Typical pattern to replace:
# BEFORE (somewhere in auth/service.ts or users/service.ts)
await sendMail({ to: user.email, subject: "...", html: "..." })

# AFTER — publish event instead
await publisher.publish("platform.auth", {
  event_id:   ulid(),
  event_type: "auth.invite_sent",
  source:     "orion-bff",
  created_at: new Date().toISOString(),
  payload: {
    recipient_ids: [user.id],
    invite_url:    inviteUrl,
    inviter_id:    actorId,
  },
})

Events to wire

Acceptance criteria

• Inviting a new Orion user in staging triggers an invite email via the notification service (check notification_log for a delivered row)
• Requesting a password reset in staging sends the reset email via the notification service (not the old SMTP path)
• grep -rn "sendMail\|nodemailer" backend/api/ returns no results after the old call is removed
• npm run verify passes in the orion repo

13NS-12 — Publish job events from Orion BFF Low risk

Goal

Add publisher.publish("platform.jobs", ...) calls in the Orion BFF service layer for the four job lifecycle events. This is additive — no existing behaviour changes.

Events to wire

Publish call pattern

await publisher.publish("platform.jobs", {
  event_id:   ulid(),
  event_type: "job.assigned",
  source:     "orion-bff",
  created_at: new Date().toISOString(),
  payload: {
    job_id:        job.jobId,
    actor_id:      actorUserId,
    recipient_ids: [job.userId],  // assigned technician
  },
})

Acceptance criteria

• Assigning a job in staging delivers an email to the technician within 10 seconds
• Changing a job status in staging delivers an email to the partner contact
• notification_log shows a row with status = sent for each triggered event
• npm run verify passes

14NS-13 — Publish invoice events from Orion BFF Low risk

Goal

Add publish calls for the three invoice lifecycle events. Identical pattern to NS-12.

Events to wire

Acceptance criteria

• Creating an invoice in staging sends an email to the partner billing contact
• Marking an invoice paid sends a confirmation email
• notification_log shows delivered rows for each invoice event in staging
• npm run verify passes

15Out of scope (deferred)