Multi-tenant SaaS

-- The tenant comes from the transaction. Unset, it is NULL, and NULL matches no row.
CREATE FUNCTION current_tenant() RETURNS bigint LANGUAGE sql STABLE
  AS $$ SELECT nullif(current_setting('app.tenant_id', true), '')::bigint $$;

ALTER TABLE projects ENABLE ROW LEVEL SECURITY;
ALTER TABLE invoices ENABLE ROW LEVEL SECURITY;
ALTER TABLE audit_log ENABLE ROW LEVEL SECURITY;
ALTER TABLE flag_overrides ENABLE ROW LEVEL SECURITY;
CREATE POLICY tenant ON projects USING (tenant_id = current_tenant());
CREATE POLICY tenant ON invoices USING (tenant_id = current_tenant());
CREATE POLICY tenant ON audit_log USING (tenant_id = current_tenant());
CREATE POLICY tenant ON flag_overrides USING (tenant_id = current_tenant());

GRANT SELECT, INSERT, UPDATE, DELETE ON projects, invoices TO tenancy_app;
GRANT SELECT ON plans, tenants, flags, flag_overrides TO tenancy_app;
-- The service can add audit rows and read them, never change them.
GRANT SELECT, INSERT ON audit_log TO tenancy_app;

-- Each request runs as the service role, with its tenant set for this transaction only.
SELECT set_config('role', 'tenancy_app', true), set_config('app.tenant_id', $1, true);

// Tick sends every subscription's head event that is due at now, one request each.
func (d *Dispatcher) Tick(ctx context.Context, now int64) (sent []Attempt, nextDue int64, err error) {
  rows, err := d.DB.Query(ctx, qw["heads"])
  if err != nil {
    return nil, 0, fmt.Errorf("read heads: %w", err)
  }
  heads, err := pgx.CollectRows(rows, func(r pgx.CollectableRow) (head, error) {
    var h head
    err := r.Scan(&h.sub, &h.seq, &h.eventID, &h.body, &h.attempts, &h.nextAt, &h.url, &h.secret)
    return h, err
  })
  if err != nil {
    return nil, 0, fmt.Errorf("read heads: %w", err)
  }
  nextDue = -1
  for _, h := range heads {
    if h.nextAt > now {
      if nextDue < 0 || h.nextAt < nextDue {
        nextDue = h.nextAt
      }
      continue
    }
    a := Attempt{At: now, Sub: h.sub, Seq: h.seq, EventID: h.eventID, N: h.attempts + 1}
    if a.Status, err = d.post(ctx, h, now); err != nil {
      return sent, 0, err
    }
    switch {
    case a.Status >= 200 && a.Status < 300:
      a.Result = "delivered"
      _, err = d.DB.Exec(ctx, qw["delivered"], h.sub, h.seq, a.Status)
    case a.N >= d.Backoff.MaxAttempts:
      a.Result = "dead"
      _, err = d.DB.Exec(ctx, qw["dead"], h.sub, h.seq, a.Status)
    default:
      a.Result = "retry"
      a.NextAt = now + d.Backoff.Delay(a.N, d.Rand)
      _, err = d.DB.Exec(ctx, qw["retry_later"], h.sub, h.seq, a.NextAt, a.Status)
      if nextDue < 0 || a.NextAt < nextDue {
        nextDue = a.NextAt
      }
    }
    if err != nil {
      return sent, 0, fmt.Errorf("save attempt %d of %s: %w", a.N, h.eventID, err)
    }
    sent = append(sent, a)
  }
  return sent, nextDue, nil
}

// post sends one signed request. It returns the status code, or 0 when no reply came in time.
func (d *Dispatcher) post(ctx context.Context, h head, now int64) (int, error) {
  body := []byte(h.body)
  req, err := http.NewRequestWithContext(ctx, http.MethodPost, h.url, bytes.NewReader(body))
  if err != nil {
    return 0, fmt.Errorf("build request for %s: %w", h.eventID, err)
  }
  req.Header.Set("Content-Type", "application/json")
  req.Header.Set("Webhook-Id", h.eventID)
  req.Header.Set(SignatureHeader, Sign(h.secret, now, body))
  resp, err := d.Client.Do(req)
  if err != nil {
    return 0, nil // a timeout or a refused connection is a failed attempt: retry later
  }
  _, copyErr := io.Copy(io.Discard, resp.Body)
  if err := errors.Join(copyErr, resp.Body.Close()); err != nil {
    return 0, fmt.Errorf("read reply for %s: %w", h.eventID, err)
  }
  return resp.StatusCode, nil
}

// Sign returns the signature header for body sent at ts: an HMAC-SHA256 over "ts.body" with
// the subscription's secret. Signing the timestamp stops a replay with a new timestamp.
func Sign(secret string, ts int64, body []byte) string {
  return fmt.Sprintf("t=%d,v1=%s", ts, digest(secret, ts, body))
}

func digest(secret string, ts int64, body []byte) string {
  mac := hmac.New(sha256.New, []byte(secret))
  fmt.Fprintf(mac, "%d.", ts)
  mac.Write(body)
  return hex.EncodeToString(mac.Sum(nil))
}

// Verify checks a signature header on the receiving side. It compares in constant time and
// refuses a timestamp more than tolerance seconds from now.
func Verify(secret, header string, body []byte, now, tolerance int64) error {
  tsPart, sigPart, ok := strings.Cut(header, ",")
  tsText, ok1 := strings.CutPrefix(tsPart, "t=")
  sig, ok2 := strings.CutPrefix(sigPart, "v1=")
  if !ok || !ok1 || !ok2 {
    return fmt.Errorf("%w: malformed header", ErrBadSignature)
  }
  ts, err := strconv.ParseInt(tsText, 10, 64)
  if err != nil {
    return fmt.Errorf("%w: bad timestamp: %w", ErrBadSignature, err)
  }
  if !hmac.Equal([]byte(digest(secret, ts, body)), []byte(sig)) {
    return ErrBadSignature
  }
  if now-ts > tolerance || ts-now > tolerance {
    return fmt.Errorf("%w: signed at %d, now %d", ErrStale, ts, now)
  }
  return nil
}

-- The head of each subscription is its oldest pending event. Later events wait behind it, so a
-- customer receives one subscription's events in order.
SELECT DISTINCT ON (d.subscription_id)
  d.subscription_id, d.seq, d.event_id, d.body, d.attempts, d.next_at, s.url, s.secret
FROM deliveries d
JOIN subscriptions s ON s.id = d.subscription_id
WHERE d.state = 'pending'
ORDER BY d.subscription_id, d.seq;