# -*- coding: utf-8 -*-
"""SQLite 상태 저장소: 파일 스냅샷, 변경 이력, (Stage 5) 확인 기록.

쓰기는 state.db 한 파일에만 한다. 감시 대상 산출물에는 절대 쓰지 않는다.
"""

from __future__ import annotations

import hmac
import json
import secrets
import sqlite3
import sys
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from pathlib import Path

from watcher.scan import FileRecord

_SCHEMA = """
CREATE TABLE IF NOT EXISTS snapshots (
    id      INTEGER PRIMARY KEY AUTOINCREMENT,
    run_at  TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS files (
    snapshot_id INTEGER NOT NULL REFERENCES snapshots(id),
    path        TEXT NOT NULL,
    sha256      TEXT NOT NULL,
    size        INTEGER NOT NULL,
    mtime       REAL NOT NULL,
    PRIMARY KEY (snapshot_id, path)
);
CREATE TABLE IF NOT EXISTS meta (
    key   TEXT PRIMARY KEY,
    value TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS extracts (
    sha256       TEXT PRIMARY KEY,
    doc_json     TEXT NOT NULL,
    extracted_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS changes (
    id          INTEGER PRIMARY KEY AUTOINCREMENT,
    snapshot_id INTEGER NOT NULL REFERENCES snapshots(id),
    kind        TEXT NOT NULL CHECK (kind IN ('added', 'modified', 'removed')),
    path        TEXT NOT NULL,
    old_sha256  TEXT,
    new_sha256  TEXT,
    detected_at TEXT NOT NULL,
    project     TEXT
);
CREATE TABLE IF NOT EXISTS issues (
    id          INTEGER PRIMARY KEY AUTOINCREMENT,
    snapshot_id INTEGER NOT NULL REFERENCES snapshots(id),
    code        TEXT NOT NULL,
    path        TEXT NOT NULL,
    project     TEXT,
    message     TEXT NOT NULL,
    evidence    TEXT NOT NULL,
    detected_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS change_details (
    change_id     INTEGER PRIMARY KEY REFERENCES changes(id),
    items_json    TEXT NOT NULL,
    captures_json TEXT
);
CREATE TABLE IF NOT EXISTS manual_pairs (
    child_key TEXT PRIMARY KEY,   -- 대상 파일 (경로 키)
    prev_key  TEXT NOT NULL,      -- 사람이 지정한 이전 판
    by        TEXT,
    at        TEXT
);
CREATE TABLE IF NOT EXISTS acks (
    id        INTEGER PRIMARY KEY AUTOINCREMENT,
    change_id INTEGER NOT NULL REFERENCES changes(id),
    by        TEXT NOT NULL,
    note      TEXT,
    at        TEXT NOT NULL
);
"""


# 변경 종류 → 한글 라벨 (콘솔·브리핑·확인 페이지 공용 — 한 곳에서만 관리)
KIND_LABEL = {"added": "추가", "modified": "수정", "removed": "삭제"}


@dataclass(frozen=True)
class Change:
    """파일 수준 변경 1건. old/new sha256이 '사실'의 근거다 (PRINCIPLE 4·5)."""
    kind: str  # 'added' | 'modified' | 'removed'
    path: str
    old_sha256: str | None
    new_sha256: str | None
    id: int | None = None       # DB에 기록된 뒤 부여 (확인 링크의 대상)
    project: str | None = None  # 귀속 후 set_change_project로 저장


class State:
    def __init__(self, db_path: Path | str):
        self.db_path = Path(db_path)
        # 웹서버(다중 스레드)와 검사 스레드가 같은 DB를 쓴다 — WAL이면 읽기가
        # 긴 쓰기(캡처 수 MB 저장)에 막히지 않고, busy_timeout으로 잠금 충돌을
        # 재시도한다 (2026-08-26 리뷰: 'database is locked' 예방)
        self.conn = sqlite3.connect(self.db_path, timeout=15)
        mode = self.conn.execute("PRAGMA journal_mode=WAL").fetchone()[0]
        if str(mode).lower() != "wal":  # 전환 실패는 예외가 아니라 반환값으로만 알린다
            print(f"경고: state.db WAL 전환 실패(현재 {mode}) — 동시 접근 시 "
                  f"잠금 오류 가능", file=sys.stderr)
        if str(self.db_path).startswith("\\\\"):
            print("경고: state.db가 네트워크 경로에 있습니다 — SQLite WAL은 "
                  "네트워크 드라이브에서 신뢰할 수 없습니다(손상 위험). "
                  "이 PC의 로컬 경로를 권장합니다.", file=sys.stderr)
        self.conn.execute("PRAGMA busy_timeout=15000")
        self.conn.execute("PRAGMA synchronous=NORMAL")
        self.conn.executescript(_SCHEMA)
        # Stage 4 이전에 만든 DB 마이그레이션: changes.project 열 추가
        cols = [r[1] for r in self.conn.execute("PRAGMA table_info(changes)")]
        if "project" not in cols:
            self.conn.execute("ALTER TABLE changes ADD COLUMN project TEXT")
        # v0.14 마이그레이션: 이슈 해소(닫기) 기록 — 없으면 빨간 숫자가 영원히 쌓인다
        icols = [r[1] for r in self.conn.execute("PRAGMA table_info(issues)")]
        if "resolved_at" not in icols:
            self.conn.execute("ALTER TABLE issues ADD COLUMN resolved_at TEXT")
            self.conn.execute("ALTER TABLE issues ADD COLUMN resolved_by TEXT")
        # v0.19 마이그레이션: 상세 항목 오탐 숨김(복원 가능) 기록
        dcols = [r[1] for r in self.conn.execute("PRAGMA table_info(change_details)")]
        if "dismissed_json" not in dcols:
            self.conn.execute(
                "ALTER TABLE change_details ADD COLUMN dismissed_json TEXT")
        # v0.21 마이그레이션: 사람이 직접 추가한 '놓친 변경' 기록 (미탐 보완)
        if "user_notes_json" not in dcols:
            self.conn.execute(
                "ALTER TABLE change_details ADD COLUMN user_notes_json TEXT")
        self.conn.commit()

    def close(self) -> None:
        self.conn.close()

    def __enter__(self) -> "State":
        return self

    def __exit__(self, *exc) -> None:
        self.close()

    def save_extract(self, sha256: str, doc_dict: dict) -> None:
        """추출 결과를 내용 해시 기준으로 저장. 나중에 구버전과의 내용 비교에 쓴다."""
        now = datetime.now(timezone.utc).isoformat()
        self.conn.execute(
            "INSERT OR REPLACE INTO extracts (sha256, doc_json, extracted_at) VALUES (?, ?, ?)",
            (sha256, json.dumps(doc_dict, ensure_ascii=False), now))
        self.conn.commit()

    def load_extract(self, sha256: str | None) -> dict | None:
        if not sha256:
            return None
        row = self.conn.execute(
            "SELECT doc_json FROM extracts WHERE sha256 = ?", (sha256,)).fetchone()
        return json.loads(row[0]) if row else None

    def latest_snapshot(self) -> dict[str, str]:
        """가장 최근 스냅샷: {path: sha256}. 없으면 빈 dict."""
        row = self.conn.execute("SELECT MAX(id) FROM snapshots").fetchone()
        if row[0] is None:
            return {}
        return dict(self.conn.execute(
            "SELECT path, sha256 FROM files WHERE snapshot_id = ?", (row[0],)))

    def _latest_rows(self) -> dict[str, tuple]:
        """최근 스냅샷의 전체 행: {path: (sha256, size, mtime)}."""
        row = self.conn.execute("SELECT MAX(id) FROM snapshots").fetchone()
        if row[0] is None:
            return {}
        return {p: (s, sz, mt) for p, s, sz, mt in self.conn.execute(
            "SELECT path, sha256, size, mtime FROM files WHERE snapshot_id = ?", (row[0],))}

    def record_run(self, records: list[FileRecord],
                   carry_paths: tuple | list = (),
                   known_prefixes: list[str] | None = None) -> tuple[int, list[Change]]:
        """이전 스냅샷과 비교해 변경을 계산하고, 새 스냅샷+변경을 저장한다.

        carry_paths: 이번 스캔에서 읽지 못한 경로들. 이전 스냅샷 값을 그대로 이월해
        '삭제'로 오인하지 않는다 (잠긴 파일이 removed로 잡히는 것 방지).
        known_prefixes: 다중 감시 폴더의 라벨 목록. 지정되면, 더 이상 감시하지 않는
        라벨 아래의 파일은 '삭제' 알림 없이 조용히 스냅샷에서 빠진다
        (감시 폴더를 목록에서 뺐을 뿐 파일이 지워진 게 아니므로).
        """
        if carry_paths:
            prev = self._latest_rows()
            have = {r.path for r in records}
            records = records + [
                FileRecord(p, *prev[p]) for p in carry_paths
                if p in prev and p not in have]
        old = self.latest_snapshot()
        new = {r.path: r for r in records}

        changes: list[Change] = []
        for path in sorted(new.keys() - old.keys()):
            changes.append(Change("added", path, None, new[path].sha256))
        for path in sorted(new.keys() & old.keys()):
            if new[path].sha256 != old[path]:
                changes.append(Change("modified", path, old[path], new[path].sha256))
        for path in sorted(old.keys() - new.keys()):
            if known_prefixes is not None and path.split("/")[0] not in known_prefixes:
                continue  # 감시 폴더 목록에서 빠진 라벨: 삭제 알림 없이 제외
            changes.append(Change("removed", path, old[path], None))

        now = datetime.now(timezone.utc).isoformat()
        cur = self.conn.execute("INSERT INTO snapshots (run_at) VALUES (?)", (now,))
        snapshot_id = cur.lastrowid
        self.conn.executemany(
            "INSERT INTO files (snapshot_id, path, sha256, size, mtime) VALUES (?, ?, ?, ?, ?)",
            [(snapshot_id, r.path, r.sha256, r.size, r.mtime) for r in records])
        with_ids = []
        for c in changes:
            cur = self.conn.execute(
                "INSERT INTO changes (snapshot_id, kind, path, old_sha256, new_sha256, detected_at) "
                "VALUES (?, ?, ?, ?, ?, ?)",
                (snapshot_id, c.kind, c.path, c.old_sha256, c.new_sha256, now))
            with_ids.append(Change(c.kind, c.path, c.old_sha256, c.new_sha256, cur.lastrowid))
        self.conn.commit()
        return snapshot_id, with_ids

    def save_change_details(self, change_id: int, items: list[dict],
                            captures: dict | None = None) -> None:
        """변경 1건의 내용 비교 결과(diff 항목 + 위치 캡처)를 화면 조회용으로 저장."""
        self.conn.execute(
            "INSERT OR REPLACE INTO change_details (change_id, items_json, captures_json)"
            " VALUES (?, ?, ?)",
            (change_id, json.dumps(items, ensure_ascii=False),
             json.dumps(captures or {}, ensure_ascii=False)))
        self.conn.commit()

    def change_detail(self, change_id: int) -> dict | None:
        row = self.conn.execute(
            "SELECT items_json, captures_json, dismissed_json, user_notes_json "
            "FROM change_details WHERE change_id = ?", (change_id,)).fetchone()
        if row is None:
            return None
        return {"items": json.loads(row[0]), "captures": json.loads(row[1] or "{}"),
                "dismissed": json.loads(row[2] or "{}"),
                "user_notes": json.loads(row[3] or "[]")}

    def add_user_note(self, change_id: int, where: str, text: str, by: str) -> bool:
        """사람이 직접 추가하는 '놓친 변경' 기록 — 프로그램이 인식 못한 변경을
        사람이 보완한다 (오탐 숨김의 반대쪽, 원칙 5 양방향 완성 — 2026-08-27).
        상세가 없는 변경(기준선 등)에는 빈 상세를 만들어서라도 붙인다."""
        row = self.conn.execute(
            "SELECT user_notes_json FROM change_details WHERE change_id = ?",
            (change_id,)).fetchone()
        if row is None:
            if self.get_change(change_id) is None:
                return False
            self.save_change_details(change_id, [], {})
            notes = []
        else:
            notes = json.loads(row[0] or "[]")
        notes.append({"where": where, "text": text, "by": by or "-",
                      "at": datetime.now(timezone.utc).isoformat()})
        self.conn.execute(
            "UPDATE change_details SET user_notes_json = ? WHERE change_id = ?",
            (json.dumps(notes, ensure_ascii=False), change_id))
        self.conn.commit()
        return True

    def delete_user_note(self, change_id: int, idx: int) -> None:
        row = self.conn.execute(
            "SELECT user_notes_json FROM change_details WHERE change_id = ?",
            (change_id,)).fetchone()
        if row is None:
            return
        notes = json.loads(row[0] or "[]")
        if 0 <= idx < len(notes):
            notes.pop(idx)
            self.conn.execute(
                "UPDATE change_details SET user_notes_json = ? WHERE change_id = ?",
                (json.dumps(notes, ensure_ascii=False), change_id))
            self.conn.commit()

    def list_user_notes(self) -> list[dict]:
        """전체 '놓친 변경' 기록 (최신순) — 어떤 파일·양식에서 프로그램이 자주
        놓치는지 보는 미탐 리포트용."""
        out = []
        for cid, path, proj, notes_json in self.conn.execute(
                "SELECT d.change_id, c.path, c.project, d.user_notes_json "
                "FROM change_details d JOIN changes c ON c.id = d.change_id "
                "WHERE d.user_notes_json IS NOT NULL"):
            for n in json.loads(notes_json or "[]"):
                out.append({**n, "change_id": cid, "path": path,
                            "project": proj or "미분류"})
        out.sort(key=lambda n: n.get("at", ""), reverse=True)
        return out

    # ------------------------------------------------------------ 수동 짝 지정
    # 자동 규칙(같은 폴더·이름)이 못 엮는 판을 사람이 직접 짝지음 — 이름 오타,
    # '이전판' 폴더 이동, 영어 꼬리 등 (2026-08-29). 원칙 5: 최종 판단은 사람.

    def set_manual_pair(self, child_key: str, prev_key: str, by: str) -> None:
        self.conn.execute(
            "INSERT OR REPLACE INTO manual_pairs (child_key, prev_key, by, at) "
            "VALUES (?, ?, ?, ?)",
            (child_key, prev_key, by or "-",
             datetime.now(timezone.utc).isoformat()))
        self.conn.commit()

    def clear_manual_pair(self, child_key: str) -> None:
        self.conn.execute("DELETE FROM manual_pairs WHERE child_key = ?",
                          (child_key,))
        self.conn.commit()

    def manual_pair(self, child_key: str) -> dict | None:
        """{prev_key, by, at} — 지정이 없으면 None."""
        row = self.conn.execute(
            "SELECT prev_key, by, at FROM manual_pairs WHERE child_key = ?",
            (child_key,)).fetchone()
        return ({"prev_key": row[0], "by": row[1], "at": row[2]}
                if row else None)

    def dismiss_change_item(self, change_id: int, idx: int, by: str) -> None:
        """상세 항목 하나를 '오탐/무의미'로 숨김 — 완전 삭제가 아니라 누가·언제와
        함께 보관하고 복원 가능 (원칙 5: 내용은 초안, 최종 판단은 사람 — 2026-08-27).
        idx는 items_json 배열 인덱스 (저장 순서 불변)."""
        row = self.conn.execute(
            "SELECT dismissed_json FROM change_details WHERE change_id = ?",
            (change_id,)).fetchone()
        if row is None:
            return
        d = json.loads(row[0] or "{}")
        d[str(idx)] = {"by": by or "-",
                       "at": datetime.now(timezone.utc).isoformat()}
        self.conn.execute(
            "UPDATE change_details SET dismissed_json = ? WHERE change_id = ?",
            (json.dumps(d, ensure_ascii=False), change_id))
        self.conn.commit()

    def restore_change_item(self, change_id: int, idx: int) -> None:
        row = self.conn.execute(
            "SELECT dismissed_json FROM change_details WHERE change_id = ?",
            (change_id,)).fetchone()
        if row is None:
            return
        d = json.loads(row[0] or "{}")
        d.pop(str(idx), None)
        self.conn.execute(
            "UPDATE change_details SET dismissed_json = ? WHERE change_id = ?",
            (json.dumps(d, ensure_ascii=False), change_id))
        self.conn.commit()

    def delete_run(self, snapshot_id: int) -> None:
        """실패한 실행의 스냅샷·변경·이슈를 되돌린다 — 다음 실행이 다시 감지하도록.
        (보고 단계 실패 시 알림 유실 방지. extracts는 내용 기준 캐시라 남겨도 무해)"""
        self.conn.execute(
            "DELETE FROM change_details WHERE change_id IN "
            "(SELECT id FROM changes WHERE snapshot_id = ?)", (snapshot_id,))
        self.conn.execute(  # 롤백된 변경의 확인 기록이 고아로 남지 않게 (2026-08-26 리뷰)
            "DELETE FROM acks WHERE change_id IN "
            "(SELECT id FROM changes WHERE snapshot_id = ?)", (snapshot_id,))
        self.conn.execute("DELETE FROM issues WHERE snapshot_id = ?", (snapshot_id,))
        self.conn.execute("DELETE FROM changes WHERE snapshot_id = ?", (snapshot_id,))
        self.conn.execute("DELETE FROM files WHERE snapshot_id = ?", (snapshot_id,))
        self.conn.execute("DELETE FROM snapshots WHERE id = ?", (snapshot_id,))
        self.conn.commit()

    def save_issues(self, snapshot_id: int, rows) -> list:
        """일관성 검사 결과 저장. rows: [(Issue, 프로젝트|None)] — 대시보드 표시용.

        미해결 불일치(도면-BOM 등)는 변경이 있는 날마다 다시 검출되므로
        동일 내용(code+path+message)은 중복 저장하지 않는다. 단, 최근 30일
        기록에 대해서만 — 전체 이력 대조로는 해결됐다가 재발한 불일치가
        영원히 묻힌다 (2026-08-25 리뷰 지적)."""
        cutoff = (datetime.now(timezone.utc) - timedelta(days=30)).isoformat()
        # 닫힌(resolved) 이슈는 대조에서 제외 — 처리했는데 재발하면 다시 보여야 한다
        existing = {(c, p, m) for c, p, m in self.conn.execute(
            "SELECT code, path, message FROM issues "
            "WHERE detected_at >= ? AND resolved_at IS NULL", (cutoff,))}
        now = datetime.now(timezone.utc).isoformat()
        fresh = [(i, proj) for i, proj in rows
                 if (i.code, i.path, i.message) not in existing]
        self.conn.executemany(
            "INSERT INTO issues (snapshot_id, code, path, project, message, evidence,"
            " detected_at) VALUES (?, ?, ?, ?, ?, ?, ?)",
            [(snapshot_id, i.code, i.path, proj, i.message,
              json.dumps(list(i.evidence), ensure_ascii=False), now)
             for i, proj in fresh])
        self.conn.commit()
        # 새로 기록된 것만 돌려준다 — 브리핑은 신규만 싣고, 미해결 누적은 화면에서
        # 본다 (같은 경고가 변경 있는 날마다 메일로 재알림되던 소음 제거, 2026-08-26)
        return fresh

    def issues(self, project: str | None = None) -> list[dict]:
        """확인 필요 이력 (최신순). 프로젝트 필터 가능."""
        sql = ("SELECT id, detected_at, code, path, project, message, evidence "
               "FROM issues WHERE resolved_at IS NULL ")
        params: tuple = ()
        if project:
            sql += "AND project = ? "
            params = (project,)
        sql += "ORDER BY id DESC"
        keys = ("issue_id", "detected_at", "code", "path", "project", "message", "evidence")
        out = []
        for row in self.conn.execute(sql, params):
            d = dict(zip(keys, row))
            d["evidence"] = json.loads(d["evidence"])
            out.append(d)
        return out

    def resolve_issue(self, issue_id: int, by: str = "") -> None:
        """이상 징후를 '처리함'으로 닫는다 — 목록·누적 숫자에서 빠진다.
        기록은 지우지 않는다 (같은 내용이 다시 검출되면 새 이슈로 다시 쌓인다)."""
        now = datetime.now(timezone.utc).isoformat()
        self.conn.execute(
            "UPDATE issues SET resolved_at = ?, resolved_by = ? WHERE id = ?",
            (now, by, issue_id))
        self.conn.commit()

    def _secret(self) -> str:
        row = self.conn.execute(
            "SELECT value FROM meta WHERE key = 'ack_secret'").fetchone()
        if row:
            return row[0]
        s = secrets.token_hex(16)
        self.conn.execute("INSERT INTO meta (key, value) VALUES ('ack_secret', ?)", (s,))
        self.conn.commit()
        return s

    def ack_token(self, change_id: int) -> str:
        """확인 링크용 토큰. 링크를 아는 사람만 확인을 저장할 수 있다 (CSRF 방지)."""
        return hmac.new(self._secret().encode(), str(change_id).encode(),
                        "sha256").hexdigest()[:16]

    def csrf_token(self) -> str:
        """웹 대시보드 폼 공용 CSRF 토큰 (설치별 고정).

        127.0.0.1 바인딩이어도 CSRF는 못 막는다 — 요청이 사용자 브라우저에서
        나가기 때문. 외부 페이지는 이 토큰을 읽을 수 없으므로 폼 위조가 차단된다."""
        return hmac.new(self._secret().encode(), b"csrf-form",
                        "sha256").hexdigest()[:32]

    def prune_details(self, days: int = 90) -> int:
        """보존 기간이 지난 무거운 기록을 지워 DB 비대화를 막는다.
        ①변경 상세(diff+캡처 JSON — 캡처 포함 시 수 MB)
        ②오래된 스냅샷의 파일 목록 — 매 검사마다 전체 파일 목록이 기록되어
          파일 1,000개당 검사 1회 약 172KB씩 영구 누적됨 (2026-08-25 확장성 검토).
          비교에 쓰는 최신 스냅샷은 기간과 무관하게 보존한다.
        변경 이력(changes)·확인(acks)·추출 캐시는 남긴다. 돌려주는 값: 지운 행 수."""
        cutoff = (datetime.now(timezone.utc) - timedelta(days=days)).isoformat()
        cur = self.conn.execute(
            "DELETE FROM change_details WHERE change_id IN "
            "(SELECT id FROM changes WHERE detected_at < ?)", (cutoff,))
        n = cur.rowcount
        cur = self.conn.execute(
            "DELETE FROM files WHERE snapshot_id IN ("
            " SELECT id FROM snapshots WHERE run_at < ?"
            " AND id != (SELECT MAX(id) FROM snapshots))", (cutoff,))
        n += cur.rowcount
        self.conn.commit()
        return n

    def set_change_project(self, change_id: int, project: str) -> None:
        self.conn.execute("UPDATE changes SET project = ? WHERE id = ?", (project, change_id))
        self.conn.commit()

    def get_change(self, change_id: int) -> Change | None:
        row = self.conn.execute(
            "SELECT kind, path, old_sha256, new_sha256, id, project FROM changes WHERE id = ?",
            (change_id,)).fetchone()
        return Change(*row) if row else None

    def add_ack(self, change_id: int, by: str, note: str) -> bool:
        """확인 기록 저장. 대상 변경이 없으면 False."""
        if self.get_change(change_id) is None:
            return False
        now = datetime.now(timezone.utc).isoformat()
        self.conn.execute("INSERT INTO acks (change_id, by, note, at) VALUES (?, ?, ?, ?)",
                          (change_id, by, note, now))
        self.conn.commit()
        return True

    def add_ack_bulk(self, change_ids: list[int], by: str, note: str) -> int:
        """여러 변경을 한 번에 확인 처리 — 기준선 수백 건 등 (2026-08-26 신입 검토:
        하나씩 이름 쳐서 확인하는 것 말고는 미확인을 줄일 방법이 없던 문제)."""
        now = datetime.now(timezone.utc).isoformat()
        self.conn.executemany(
            "INSERT INTO acks (change_id, by, note, at) VALUES (?, ?, ?, ?)",
            [(cid, by, note, now) for cid in change_ids])
        self.conn.commit()
        return len(change_ids)

    def first_snapshot_id(self) -> int | None:
        """기준선(첫 검사) 스냅샷 id — 그때의 '추가' 대량 건은 실제 변경이 아니다."""
        row = self.conn.execute("SELECT MIN(id) FROM snapshots").fetchone()
        return row[0]

    def history(self, project: str | None = None) -> list[dict]:
        """변경 이력 + 확인 기록 (프로젝트 필터 가능)."""
        sql = ("SELECT c.id, c.detected_at, c.kind, c.path, c.project, a.by, a.note, a.at, "
               "c.snapshot_id "
               "FROM changes c LEFT JOIN acks a ON a.change_id = c.id ")
        params: tuple = ()
        if project:
            sql += "WHERE c.project = ? "
            params = (project,)
        sql += "ORDER BY c.id, a.id"
        keys = ("change_id", "detected_at", "kind", "path", "project",
                "ack_by", "ack_note", "ack_at", "snapshot_id")
        return [dict(zip(keys, row)) for row in self.conn.execute(sql, params)]
