-- 012: WP8 Live Source Monitor — per-source reducer state + observation dedupe.
-- (010/011 are reserved for the WP7 egress package; the monitor consumes those
-- budgets later and deliberately carries no rate machinery of its own.)
--
-- One row per monitored source identity (connection_id, remote_ref): the keyed
-- partition the MonitorObservation reducer folds into (proposal §4.4). Each
-- child row carries a denormalised workspace_id so RLS stays a direct column
-- predicate and composite FKs pin the row to ONE workspace's connection.
--
-- `version` is the per-source compare-and-swap token: every applied mutation
-- bumps it, updates are guarded by `WHERE ... AND version = $n`, conflicting
-- writers reload and re-evaluate, and downstream consumers (WS fan-out,
-- outbox) drop non-increasing versions.

CREATE TABLE source_monitor_states (
  id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
  workspace_id uuid NOT NULL,
  connection_id uuid NOT NULL,
  remote_ref text NOT NULL CHECK (btrim(remote_ref) <> ''),
  state text NOT NULL DEFAULT 'OFFLINE'
    CHECK (state IN ('OFFLINE', 'PROBING', 'LIVE', 'UNSTABLE', 'ERROR')),
  version bigint NOT NULL DEFAULT 0 CHECK (version >= 0),
  -- Highest *available* signal tier; NULL until the first fresh signal lands.
  authoritative_tier text
    CHECK (authoritative_tier IN ('media-health', 'hls-probe', 'webhook', 'api-poll')),
  -- Last applied observedAt per signal tier. Staleness is judged WITHIN one
  -- tier only (provider clocks are not comparable across signal sources).
  last_media_health_at timestamptz,
  last_hls_probe_at timestamptz,
  last_webhook_at timestamptz,
  last_api_poll_at timestamptz,
  -- Remaining per-tier reducer bookkeeping: lastOutcome, lastProviderEventId,
  -- deniedUntil, failStreak. jsonb keeps the reducer state shape in one place.
  tier_status jsonb NOT NULL DEFAULT '{}'::jsonb,
  ok_streak int NOT NULL DEFAULT 0 CHECK (ok_streak >= 0),
  fail_streak int NOT NULL DEFAULT 0 CHECK (fail_streak >= 0),
  state_entered_at timestamptz NOT NULL DEFAULT now(),
  -- Hysteresis: min-hold keeps a freshly entered state from being cut down,
  -- cooldown blocks the reverse transition (no rapid A -> B -> A).
  hold_until timestamptz,
  cooldown_until timestamptz,
  previous_state text CHECK (previous_state IN ('OFFLINE', 'PROBING', 'LIVE', 'UNSTABLE', 'ERROR')),
  last_transition_at timestamptz,
  last_transition_reason text,
  created_at timestamptz NOT NULL DEFAULT now(),
  updated_at timestamptz NOT NULL DEFAULT now(),
  UNIQUE (workspace_id, connection_id, remote_ref),
  UNIQUE (id, workspace_id),
  FOREIGN KEY (connection_id, workspace_id) REFERENCES connections (id, workspace_id) ON DELETE CASCADE
);

CREATE INDEX source_monitor_states_ws_created
  ON source_monitor_states (workspace_id, created_at DESC, id DESC);
CREATE INDEX source_monitor_states_ws_state
  ON source_monitor_states (workspace_id, state);

-- Observation dedupe is enforced by the DB, not the app (proposal §4.4): a
-- replayed providerEventId can never apply twice, regardless of which process
-- or replica delivers it. Rows are journal entries, never mutated.
CREATE TABLE monitor_seen_events (
  workspace_id uuid NOT NULL,
  provider_event_id text NOT NULL CHECK (btrim(provider_event_id) <> ''),
  source_monitor_state_id uuid NOT NULL,
  seen_at timestamptz NOT NULL DEFAULT now(),
  PRIMARY KEY (workspace_id, provider_event_id),
  FOREIGN KEY (source_monitor_state_id, workspace_id)
    REFERENCES source_monitor_states (id, workspace_id) ON DELETE CASCADE
);
CREATE INDEX monitor_seen_events_seen_at ON monitor_seen_events (seen_at);

CREATE TRIGGER trg_monitor_state_ws_immutable BEFORE UPDATE ON source_monitor_states
  FOR EACH ROW EXECUTE FUNCTION reject_workspace_change();
CREATE TRIGGER trg_monitor_seen_event_ws_immutable BEFORE UPDATE ON monitor_seen_events
  FOR EACH ROW EXECUTE FUNCTION reject_workspace_change();

ALTER TABLE source_monitor_states ENABLE ROW LEVEL SECURITY;
ALTER TABLE monitor_seen_events ENABLE ROW LEVEL SECURITY;
ALTER TABLE source_monitor_states FORCE ROW LEVEL SECURITY;
ALTER TABLE monitor_seen_events FORCE ROW LEVEL SECURITY;

CREATE POLICY tenant_isolation ON source_monitor_states
  USING (workspace_id = NULLIF(current_setting('app.workspace_id', true), '')::uuid)
  WITH CHECK (workspace_id = NULLIF(current_setting('app.workspace_id', true), '')::uuid);
CREATE POLICY tenant_isolation ON monitor_seen_events
  USING (workspace_id = NULLIF(current_setting('app.workspace_id', true), '')::uuid)
  WITH CHECK (workspace_id = NULLIF(current_setting('app.workspace_id', true), '')::uuid);

GRANT SELECT, INSERT, UPDATE, DELETE ON source_monitor_states, monitor_seen_events TO sp_app;

-- Narrow SECURITY DEFINER lookup: the monitor/scheduler processes discover the
-- tenant that owns a source without weakening FORCE RLS for ordinary access
-- (same standing as publishing_schedule_workspace in 006).
CREATE FUNCTION live_monitor_source_workspace(p_connection_id uuid, p_remote_ref text) RETURNS uuid
LANGUAGE sql STABLE SECURITY DEFINER SET search_path = public AS $$
  SELECT workspace_id
  FROM source_monitor_states
  WHERE connection_id = p_connection_id AND remote_ref = p_remote_ref
  LIMIT 1
$$;

-- Retention for the dedupe journal. Runs in system context on purpose: pruned
-- rows span tenants, and a tenant that stops recording must not leak journal
-- rows forever.
CREATE FUNCTION monitor_prune_seen_events(p_before timestamptz) RETURNS integer
LANGUAGE sql SECURITY DEFINER SET search_path = public AS $$
  WITH removed AS (
    DELETE FROM monitor_seen_events WHERE seen_at < p_before RETURNING 1
  )
  SELECT count(*) FROM removed
$$;

REVOKE ALL ON FUNCTION live_monitor_source_workspace(uuid, text) FROM PUBLIC;
REVOKE ALL ON FUNCTION monitor_prune_seen_events(timestamptz) FROM PUBLIC;
GRANT EXECUTE ON FUNCTION live_monitor_source_workspace(uuid, text), monitor_prune_seen_events(timestamptz) TO sp_app;