DROP VIEW IF EXISTS moraine.mv_search_documents_from_normalized;
DROP VIEW IF EXISTS moraine.mv_search_documents_from_compacted;
DROP VIEW IF EXISTS moraine.mv_search_documents_from_events;
DROP VIEW IF EXISTS moraine.mv_search_postings;
DROP VIEW IF EXISTS moraine.mv_search_term_stats;
DROP VIEW IF EXISTS moraine.mv_search_corpus_stats;

CREATE TABLE IF NOT EXISTS moraine.search_documents (
  doc_version UInt64,
  ingested_at DateTime64(3),
  event_uid String,
  compacted_parent_uid String,
  session_id String,
  session_date Date,
  source_name LowCardinality(String) DEFAULT '',
  harness LowCardinality(String) DEFAULT '',
  inference_provider LowCardinality(String) DEFAULT '',
  endpoint_kind LowCardinality(String) DEFAULT 'generation',
  source_file String,
  source_generation UInt32,
  source_line_no UInt64,
  source_offset UInt64,
  source_ref String,
  record_ts String,
  event_class LowCardinality(String),
  payload_type LowCardinality(String),
  actor_role LowCardinality(String),
  name LowCardinality(String),
  phase LowCardinality(String),
  text_content String,
  payload_json String,
  token_usage_json String,
  token_usage_buckets Map(String, UInt64) DEFAULT map(
    'input_text', toUInt64(0),
    'output_text', toUInt64(0),
    'input_cache_read', toUInt64(0),
    'input_cache_write', toUInt64(0),
    'input_image', toUInt64(0),
    'output_image', toUInt64(0),
    'input_audio', toUInt64(0),
    'output_audio', toUInt64(0),
    'reasoning', toUInt64(0),
    'server_tool_use', toUInt64(0),
    'embedding_input_text', toUInt64(0),
    'embedding_input_image', toUInt64(0),
    'other', toUInt64(0)
  ),
  token_usage_native_units Map(String, Float64) DEFAULT map(
    'input_image_pixels', toFloat64(0),
    'output_image_pixels', toFloat64(0),
    'input_audio_seconds', toFloat64(0),
    'output_audio_seconds', toFloat64(0),
    'input_images', toFloat64(0),
    'output_images', toFloat64(0)
  ),
  doc_len UInt32 MATERIALIZED toUInt32(length(extractAll(lowerUTF8(text_content), '[a-z0-9_]+')))
)
ENGINE = ReplacingMergeTree(doc_version)
PARTITION BY toYYYYMM(ingested_at)
ORDER BY (event_uid);

ALTER TABLE moraine.search_documents
  ADD COLUMN IF NOT EXISTS source_name LowCardinality(String) DEFAULT '' AFTER session_date;
ALTER TABLE moraine.search_documents
  ADD COLUMN IF NOT EXISTS harness LowCardinality(String) DEFAULT '' AFTER source_name;
ALTER TABLE moraine.search_documents
  ADD COLUMN IF NOT EXISTS inference_provider LowCardinality(String) DEFAULT '' AFTER harness;
ALTER TABLE moraine.search_documents
  ADD COLUMN IF NOT EXISTS endpoint_kind LowCardinality(String) DEFAULT 'generation' AFTER inference_provider;
ALTER TABLE moraine.search_documents
  ADD COLUMN IF NOT EXISTS token_usage_buckets Map(String, UInt64) DEFAULT map(
    'input_text', toUInt64(0),
    'output_text', toUInt64(0),
    'input_cache_read', toUInt64(0),
    'input_cache_write', toUInt64(0),
    'input_image', toUInt64(0),
    'output_image', toUInt64(0),
    'input_audio', toUInt64(0),
    'output_audio', toUInt64(0),
    'reasoning', toUInt64(0),
    'server_tool_use', toUInt64(0),
    'embedding_input_text', toUInt64(0),
    'embedding_input_image', toUInt64(0),
    'other', toUInt64(0)
  ) AFTER token_usage_json;
ALTER TABLE moraine.search_documents
  ADD COLUMN IF NOT EXISTS token_usage_native_units Map(String, Float64) DEFAULT map(
    'input_image_pixels', toFloat64(0),
    'output_image_pixels', toFloat64(0),
    'input_audio_seconds', toFloat64(0),
    'output_audio_seconds', toFloat64(0),
    'input_images', toFloat64(0),
    'output_images', toFloat64(0)
  ) AFTER token_usage_buckets;

CREATE MATERIALIZED VIEW IF NOT EXISTS moraine.mv_search_documents_from_events
TO moraine.search_documents
AS
SELECT
  event_version AS doc_version,
  ingested_at,
  event_uid,
  origin_event_id AS compacted_parent_uid,
  session_id,
  session_date,
  source_name,
  harness,
  inference_provider,
  endpoint_kind,
  source_file,
  source_generation,
  source_line_no,
  source_offset,
  source_ref,
  record_ts,
  event_kind AS event_class,
  payload_type,
  actor_kind AS actor_role,
  tool_name AS name,
  if(tool_phase != '', tool_phase, op_status) AS phase,
  text_content,
  payload_json,
  token_usage_json,
  token_usage_buckets,
  token_usage_native_units
FROM moraine.events
WHERE lengthUTF8(replaceRegexpAll(text_content, '\\s+', '')) > 0;

CREATE TABLE IF NOT EXISTS moraine.search_postings (
  post_version UInt64,
  term String,
  doc_id String,
  session_id String,
  source_name LowCardinality(String) DEFAULT '',
  harness LowCardinality(String) DEFAULT '',
  inference_provider LowCardinality(String) DEFAULT '',
  event_class LowCardinality(String),
  payload_type LowCardinality(String),
  actor_role LowCardinality(String),
  name LowCardinality(String),
  phase LowCardinality(String),
  source_ref String,
  doc_len UInt32,
  tf UInt16
)
ENGINE = ReplacingMergeTree(post_version)
PARTITION BY cityHash64(term) % 32
ORDER BY (term, doc_id);

ALTER TABLE moraine.search_postings
  ADD COLUMN IF NOT EXISTS source_name LowCardinality(String) DEFAULT '' AFTER session_id;
ALTER TABLE moraine.search_postings
  ADD COLUMN IF NOT EXISTS harness LowCardinality(String) DEFAULT '' AFTER source_name;
ALTER TABLE moraine.search_postings
  ADD COLUMN IF NOT EXISTS inference_provider LowCardinality(String) DEFAULT '' AFTER harness;

CREATE MATERIALIZED VIEW IF NOT EXISTS moraine.mv_search_postings
TO moraine.search_postings
AS
SELECT
  d.doc_version AS post_version,
  d.term,
  d.event_uid AS doc_id,
  d.session_id,
  d.source_name,
  d.harness,
  d.inference_provider,
  d.event_class,
  d.payload_type,
  d.actor_role,
  d.name,
  d.phase,
  d.source_ref,
  d.doc_len,
  toUInt16(count()) AS tf
FROM
(
  SELECT
    doc_version,
    event_uid,
    session_id,
    source_name,
    harness,
    inference_provider,
    event_class,
    payload_type,
    actor_role,
    name,
    phase,
    source_ref,
    doc_len,
    arrayJoin(extractAll(lowerUTF8(text_content), '[a-z0-9_]+')) AS term
  FROM moraine.search_documents
  WHERE doc_len > 0
) AS d
WHERE lengthUTF8(d.term) BETWEEN 2 AND 64
GROUP BY
  d.doc_version,
  d.term,
  d.event_uid,
  d.session_id,
  d.source_name,
  d.harness,
  d.inference_provider,
  d.event_class,
  d.payload_type,
  d.actor_role,
  d.name,
  d.phase,
  d.source_ref,
  d.doc_len;

DROP TABLE IF EXISTS moraine.search_term_stats;

CREATE VIEW IF NOT EXISTS moraine.search_term_stats
AS
SELECT
  term,
  toUInt64(count()) AS docs
FROM moraine.search_postings FINAL
GROUP BY term;

DROP TABLE IF EXISTS moraine.search_corpus_stats;

CREATE VIEW IF NOT EXISTS moraine.search_corpus_stats
AS
SELECT
  toUInt8(0) AS bucket,
  toUInt64(count()) AS docs,
  toUInt64(ifNull(sum(doc_len), 0)) AS total_doc_len
FROM moraine.search_documents FINAL
WHERE doc_len > 0;

CREATE TABLE IF NOT EXISTS moraine.search_query_log (
  ts DateTime64(3) DEFAULT now64(3),
  query_id String,
  source LowCardinality(String),
  session_hint String,
  raw_query String,
  normalized_terms Array(String),
  term_count UInt16,
  result_limit UInt16,
  min_should_match UInt16,
  min_score Float64,
  include_tool_events UInt8,
  exclude_codex_mcp UInt8,
  response_ms UInt32,
  result_count UInt16,
  metadata_json String
)
ENGINE = MergeTree
PARTITION BY toYYYYMM(ts)
ORDER BY (ts, query_id);

CREATE TABLE IF NOT EXISTS moraine.search_hit_log (
  ts DateTime64(3) DEFAULT now64(3),
  query_id String,
  rank UInt16,
  event_uid String,
  session_id String,
  source_name LowCardinality(String) DEFAULT '',
  harness LowCardinality(String) DEFAULT '',
  inference_provider LowCardinality(String) DEFAULT '',
  score Float64,
  matched_terms UInt16,
  doc_len UInt32,
  event_class LowCardinality(String),
  payload_type LowCardinality(String),
  actor_role LowCardinality(String),
  name LowCardinality(String),
  source_ref String
)
ENGINE = MergeTree
PARTITION BY toYYYYMM(ts)
ORDER BY (query_id, rank, ts);

ALTER TABLE moraine.search_hit_log
  ADD COLUMN IF NOT EXISTS source_name LowCardinality(String) DEFAULT '' AFTER session_id;
ALTER TABLE moraine.search_hit_log
  ADD COLUMN IF NOT EXISTS harness LowCardinality(String) DEFAULT '' AFTER source_name;
ALTER TABLE moraine.search_hit_log
  ADD COLUMN IF NOT EXISTS inference_provider LowCardinality(String) DEFAULT '' AFTER harness;

CREATE TABLE IF NOT EXISTS moraine.search_interaction_log (
  ts DateTime64(3) DEFAULT now64(3),
  interaction_id String,
  query_id String,
  event_uid String,
  action LowCardinality(String),
  actor String,
  value Float64,
  note String,
  metadata_json String
)
ENGINE = MergeTree
PARTITION BY toYYYYMM(ts)
ORDER BY (ts, query_id, event_uid, action);
