wulf-pulse/migrations/079_endpoint_data_model.sql

-- Endpoint data model — anchor on configuration_items, sidecar everything else.
--
-- Adds three tables:
--   1. device_external_ids — cross-reference: one row per (tool, device-in-tool).
--   2. device_observations — time-series script/telemetry results (JSONB),
--      with optional B2 pointer for blobs that don't belong inline.
--   3. endpoint_audits     — LLM-produced audits anchored to a configuration_item.
--      Replaces itglue_asset_audits rows where asset_type='configuration'.
--      Flexible-asset audits stay in itglue_asset_audits (different lifecycle).
--
-- Backfill strategy is conservative: write source rows for every tool, but
-- only auto-link to a configuration_item when the match is deterministic
-- (Autotask's own rmm_device_uid for Datto). Hostname/serial fuzzy linking
-- is the reconciliation cron's job — it logs conflicts instead of merging.

-- ---------------------------------------------------------------------------
-- 1. device_external_ids
-- ---------------------------------------------------------------------------

CREATE TABLE IF NOT EXISTS device_external_ids (
  id                    BIGSERIAL PRIMARY KEY,
  configuration_item_id BIGINT NULL REFERENCES configuration_items(id) ON DELETE SET NULL,
  source                TEXT   NOT NULL,
  source_id             TEXT   NOT NULL,
  hostname              TEXT,
  serial                TEXT,
  mac                   TEXT,
  -- No FK on company_id: configuration_items already tolerates dangling
  -- company_ids (Autotask retains CIs after company hard-deletes), and the
  -- xref shouldn't be stricter than its anchor.
  company_id            BIGINT NULL,
  last_seen_at          TIMESTAMPTZ,
  link_confidence       TEXT,
  linked_at             TIMESTAMPTZ,
  created_at            TIMESTAMPTZ NOT NULL DEFAULT NOW(),
  updated_at            TIMESTAMPTZ NOT NULL DEFAULT NOW(),
  CONSTRAINT device_external_ids_source_check
    CHECK (source IN ('autotask','datto_rmm','itglue','s1','veeam','addigy','auvik')),
  CONSTRAINT device_external_ids_link_confidence_check
    CHECK (link_confidence IS NULL OR link_confidence IN
      ('canonical','exact_uid','exact_serial','hostname_in_company','mac','manual')),
  CONSTRAINT device_external_ids_unique UNIQUE (source, source_id)
);

CREATE INDEX IF NOT EXISTS ix_device_external_ids_ci
  ON device_external_ids(configuration_item_id) WHERE configuration_item_id IS NOT NULL;
CREATE INDEX IF NOT EXISTS ix_device_external_ids_unlinked
  ON device_external_ids(source, last_seen_at) WHERE configuration_item_id IS NULL;
CREATE INDEX IF NOT EXISTS ix_device_external_ids_hostname
  ON device_external_ids(LOWER(hostname)) WHERE hostname IS NOT NULL;
CREATE INDEX IF NOT EXISTS ix_device_external_ids_serial
  ON device_external_ids(serial) WHERE serial IS NOT NULL;

COMMENT ON TABLE device_external_ids IS
  'Cross-reference from per-tool device IDs to a canonical Autotask configuration_item. configuration_item_id NULL = seen in tool X but no Autotask CI matched yet; reconciliation cron tries to link.';

-- ---------------------------------------------------------------------------
-- 2. device_observations
-- ---------------------------------------------------------------------------

CREATE TABLE IF NOT EXISTS device_observations (
  id                    UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  configuration_item_id BIGINT NULL REFERENCES configuration_items(id) ON DELETE SET NULL,
  source                TEXT   NOT NULL,
  kind                  TEXT   NOT NULL,
  collected_at          TIMESTAMPTZ NOT NULL,
  payload               JSONB  NOT NULL,
  evidence_object_key   TEXT,
  run_id                TEXT,
  supersedes_id         UUID NULL REFERENCES device_observations(id) ON DELETE SET NULL,
  created_at            TIMESTAMPTZ NOT NULL DEFAULT NOW(),
  CONSTRAINT device_observations_source_check
    CHECK (source IN ('autotask','datto_rmm','itglue','s1','veeam','addigy','auvik','loglift','overshell','manual'))
);

CREATE INDEX IF NOT EXISTS ix_device_observations_ci_kind_time
  ON device_observations(configuration_item_id, kind, collected_at DESC)
  WHERE configuration_item_id IS NOT NULL;
CREATE INDEX IF NOT EXISTS ix_device_observations_kind_time
  ON device_observations(kind, collected_at DESC);
CREATE INDEX IF NOT EXISTS ix_device_observations_run_id
  ON device_observations(run_id) WHERE run_id IS NOT NULL;

COMMENT ON TABLE device_observations IS
  'Time-series telemetry / script results per endpoint. Payload is JSONB for direct querying; large blobs (raw event-log gzips, screenshots) live in B2 referenced by evidence_object_key.';

-- ---------------------------------------------------------------------------
-- 3. endpoint_audits
-- ---------------------------------------------------------------------------

-- An audit anchors on an Autotask configuration_item when one is known, but
-- IT Glue is also a valid anchor for devices we track there but not in
-- Autotask (orphans, IT-Glue-first onboarding). At least one must be set —
-- the CHECK below enforces it.
CREATE TABLE IF NOT EXISTS endpoint_audits (
  id                            UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  configuration_item_id         BIGINT NULL REFERENCES configuration_items(id) ON DELETE SET NULL,
  itglue_configuration_id       BIGINT NULL,
  organization_id               BIGINT NULL,
  generated_by_user_id          TEXT NULL REFERENCES "user"(id) ON DELETE SET NULL,
  generated_at                  TIMESTAMPTZ NOT NULL DEFAULT NOW(),
  provider                      TEXT NOT NULL DEFAULT 'anthropic',
  model_used                    TEXT,
  asset_snapshot                JSONB NOT NULL,
  observations_consumed         UUID[] NOT NULL DEFAULT '{}',
  ticket_count                  INTEGER NOT NULL DEFAULT 0,
  field_gaps                    JSONB NOT NULL DEFAULT '[]'::jsonb,
  notes_promotions              JSONB NOT NULL DEFAULT '[]'::jsonb,
  contradictions                JSONB NOT NULL DEFAULT '[]'::jsonb,
  overall_score                 NUMERIC(3,2),
  estimated_cost_usd            NUMERIC(10,4),
  total_input_tokens            INTEGER,
  total_output_tokens           INTEGER,
  status                        TEXT NOT NULL DEFAULT 'complete',
  error_message                 TEXT,
  triggered_by_ticket_number    TEXT,
  triggered_by_analysis_id      UUID NULL REFERENCES analyzer_analyses(id) ON DELETE SET NULL,
  triggered_by_observation_id   UUID NULL REFERENCES device_observations(id) ON DELETE SET NULL,
  legacy_itglue_audit_id        UUID NULL,
  CONSTRAINT endpoint_audits_provider_check
    CHECK (provider IN ('anthropic','openrouter')),
  CONSTRAINT endpoint_audits_status_check
    CHECK (status IN ('pending','running','complete','failed')),
  CONSTRAINT endpoint_audits_anchor_check
    CHECK (configuration_item_id IS NOT NULL OR itglue_configuration_id IS NOT NULL)
);

-- Idempotent guard for re-runs against an already-created table.
ALTER TABLE endpoint_audits ADD COLUMN IF NOT EXISTS itglue_configuration_id BIGINT NULL;
DO $$
BEGIN
  IF NOT EXISTS (
    SELECT 1 FROM pg_constraint WHERE conname = 'endpoint_audits_anchor_check'
  ) THEN
    ALTER TABLE endpoint_audits
      ADD CONSTRAINT endpoint_audits_anchor_check
      CHECK (configuration_item_id IS NOT NULL OR itglue_configuration_id IS NOT NULL);
  END IF;
END $$;

CREATE INDEX IF NOT EXISTS ix_endpoint_audits_ci_time
  ON endpoint_audits(configuration_item_id, generated_at DESC) WHERE configuration_item_id IS NOT NULL;
CREATE INDEX IF NOT EXISTS ix_endpoint_audits_org_time
  ON endpoint_audits(organization_id, generated_at DESC) WHERE organization_id IS NOT NULL;
CREATE INDEX IF NOT EXISTS ix_endpoint_audits_ticket
  ON endpoint_audits(triggered_by_ticket_number) WHERE triggered_by_ticket_number IS NOT NULL;
CREATE INDEX IF NOT EXISTS ix_endpoint_audits_legacy
  ON endpoint_audits(legacy_itglue_audit_id) WHERE legacy_itglue_audit_id IS NOT NULL;
CREATE INDEX IF NOT EXISTS ix_endpoint_audits_itglue_config
  ON endpoint_audits(itglue_configuration_id, generated_at DESC) WHERE itglue_configuration_id IS NOT NULL;

COMMENT ON TABLE endpoint_audits IS
  'LLM-produced audits anchored to a configuration_item. The configuration variant of itglue_asset_audits — flexible-asset audits stay in itglue_asset_audits since their lifecycle is different.';
COMMENT ON COLUMN endpoint_audits.legacy_itglue_audit_id IS
  'Backfill pointer to the original itglue_asset_audits row this was migrated from. Drop after cutover is complete.';

-- ---------------------------------------------------------------------------
-- Backfill: device_external_ids
-- ---------------------------------------------------------------------------

-- Autotask: every CI is its own canonical row (the anchor).
INSERT INTO device_external_ids
  (configuration_item_id, source, source_id, hostname, serial, mac, company_id,
   last_seen_at, link_confidence, linked_at)
SELECT
  ci.id,
  'autotask',
  ci.id::text,
  ci.reference_title,
  ci.serial_number,
  ci.rmm_device_audit_mac_address,
  ci.company_id,
  ci.synced_at AT TIME ZONE 'UTC',
  'canonical',
  NOW()
FROM configuration_items ci
WHERE NOT EXISTS (
  SELECT 1 FROM device_external_ids dx
   WHERE dx.source = 'autotask' AND dx.source_id = ci.id::text
)
ON CONFLICT (source, source_id) DO NOTHING;

-- Datto RMM: link via Autotask's stored rmm_device_uid (deterministic).
-- Rows that don't match stay unlinked; reconciliation cron will retry by hostname.
INSERT INTO device_external_ids
  (configuration_item_id, source, source_id, hostname, serial, mac, company_id,
   last_seen_at, link_confidence, linked_at)
SELECT
  ci.id,
  'datto_rmm',
  drd.uid,
  drd.hostname,
  NULL,
  NULL,
  ci.company_id,
  drd.last_seen,
  CASE WHEN ci.id IS NOT NULL THEN 'exact_uid' ELSE NULL END,
  CASE WHEN ci.id IS NOT NULL THEN NOW() ELSE NULL END
FROM datto_rmm_devices drd
LEFT JOIN configuration_items ci ON ci.rmm_device_uid = drd.uid
ON CONFLICT (source, source_id) DO NOTHING;

-- IT Glue: stays unlinked initially. Reconciliation matches by serial → hostname.
INSERT INTO device_external_ids
  (source, source_id, hostname, serial, mac, last_seen_at)
SELECT
  'itglue',
  igc.id::text,
  COALESCE(igc.hostname, igc.name),
  igc.serial_number,
  igc.mac_address,
  igc.synced_at
FROM itg_configurations igc
ON CONFLICT (source, source_id) DO NOTHING;

-- SentinelOne: stays unlinked initially. Reconciliation matches by computer_name + company.
INSERT INTO device_external_ids
  (source, source_id, hostname, last_seen_at)
SELECT
  's1',
  sa.id::text,
  sa.computer_name,
  sa.last_active_date AT TIME ZONE 'UTC'
FROM s1_agents sa
ON CONFLICT (source, source_id) DO NOTHING;

-- Veeam: stays unlinked initially. Reconciliation matches by hostname (the
-- agent's `name` field). No serial / MAC available.
INSERT INTO device_external_ids
  (source, source_id, hostname, last_seen_at)
SELECT
  'veeam',
  vba.instance_uid,
  vba.name,
  vba.synced_at
FROM veeam_backup_agents vba
ON CONFLICT (source, source_id) DO NOTHING;

-- ---------------------------------------------------------------------------
-- Backfill: endpoint_audits from itglue_asset_audits (configuration only)
-- ---------------------------------------------------------------------------

-- We snapshot the configuration audits into endpoint_audits but keep the
-- originals in itglue_asset_audits until the receiver flips. legacy_itglue_audit_id
-- preserves the link both ways during the dual-source window.

-- itglue_asset_audits.asset_id (when asset_type='configuration') is an IT Glue
-- Configuration ID, not an Autotask CI. We anchor primarily on IT Glue here
-- and let the reconciler later fill in configuration_item_id via the xref.
INSERT INTO endpoint_audits (
  configuration_item_id, itglue_configuration_id,
  organization_id, generated_by_user_id, generated_at,
  provider, model_used, asset_snapshot, ticket_count,
  field_gaps, notes_promotions, contradictions,
  overall_score, estimated_cost_usd, total_input_tokens, total_output_tokens,
  status, error_message, triggered_by_ticket_number, triggered_by_analysis_id,
  legacy_itglue_audit_id
)
SELECT
  dx.configuration_item_id,
  ia.asset_id,
  ia.organization_id,
  ia.generated_by_user_id,
  ia.generated_at,
  ia.provider,
  ia.model_used,
  ia.asset_snapshot,
  ia.ticket_count,
  ia.field_gaps,
  ia.notes_promotions,
  ia.contradictions,
  ia.overall_score,
  ia.estimated_cost_usd,
  ia.total_input_tokens,
  ia.total_output_tokens,
  ia.status,
  ia.error_message,
  ia.triggered_by_ticket_number,
  ia.triggered_by_analysis_id,
  ia.id
FROM itglue_asset_audits ia
LEFT JOIN device_external_ids dx
  ON dx.source = 'itglue' AND dx.source_id = ia.asset_id::text
WHERE ia.asset_type = 'configuration'
  AND NOT EXISTS (
    SELECT 1 FROM endpoint_audits ea WHERE ea.legacy_itglue_audit_id = ia.id
  );

-- ---------------------------------------------------------------------------
-- updated_at trigger for device_external_ids
-- ---------------------------------------------------------------------------

CREATE OR REPLACE FUNCTION device_external_ids_set_updated_at() RETURNS TRIGGER AS $$
BEGIN
  NEW.updated_at = NOW();
  RETURN NEW;
END;
$$ LANGUAGE plpgsql;

DROP TRIGGER IF EXISTS trg_device_external_ids_updated_at ON device_external_ids;
CREATE TRIGGER trg_device_external_ids_updated_at
  BEFORE UPDATE ON device_external_ids
  FOR EACH ROW EXECUTE FUNCTION device_external_ids_set_updated_at();