-*- mode: org -*-
#+TITLE:       sisudoc spine (doc_reform) output pod source sqlite db
#+DESCRIPTION: documents - structuring, publishing in multiple formats & search
#+FILETAGS:    :spine:output:source:pod:
#+AUTHOR:      Ralph Amissah
#+EMAIL:       [[mailto:ralph.amissah@gmail.com][ralph.amissah@gmail.com]]
#+COPYRIGHT:   Copyright (C) 2015 (continuously updated, current 2026) Ralph Amissah
#+LANGUAGE:    en
#+STARTUP:     content hideblocks hidestars noindent entitiespretty
#+PROPERTY:    header-args+ :eval never-export :exports code
#+PROPERTY:    header-args+ :noweb yes :padline no
#+PROPERTY:    header-args+ :results silent :cache no
#+PROPERTY:    header-args+ :mkdirp yes
#+OPTIONS:     H:3 num:nil toc:t \n:t ::t |:t ^:nil -:t f:t *:t
- magic single double-quote → " ← FIX changes hilighting behavior (occuring
  after it) in org document. INVESTIGATE (org-mode CONFIG?) FIND & FIX

- [[./doc-reform.org][doc-reform.org]]  [[./][org/]]

* (Object-Centric) Document Abstraction SQLite db

- Process markup document, create document abstraction

** _module template_ :module:metadoc_from_src:

rename source_abstraction_peg_txt.d

#+HEADER: :tangle "../src/sisudoc/outputs/io_out/sqlite_ocda_db.d"
#+HEADER: :noweb yes
#+BEGIN_SRC d
<<doc_header_including_copyright_and_license>>
module sisudoc.outputs.io_out.sqlite_ocda_db;

/+ ↓ write document abstraction as per-document sqlite3 database +/
template spineAbstractionDb() {
  import std.conv : to;
  import std.digest : toHexString;
  import std.file;
  import std.path;
  import std.stdio;
  import std.string;
  import std.array;
  import d2sqlite3;
  import sisudoc.outputs.io_out.paths_output;
  import sisudoc.ocda.io_in.paths_source;
  import sisudoc.ocda.abstraction.ssp : ssp_format_version;
  import sisudoc.ocda.zstd : zstdCompress;
  /+ ↓ the document is passed in rather than taken from doc, because the
       caller hands over one that has been through the .ssp: written as text
       and read back. so the database is a function of the .ssp and cannot
       carry anything the .ssp does not, header blocks included: the metadata
       table below is filled from what came back, not from a second list read
       off doc_matters, which is how the two drifted apart before
  +/
  void spineAbstractionDb(D,S)(D doc, S ssp_doc) {
    auto doc_matters     = doc.matters;
    auto doc_abstraction = ssp_doc.abstraction;
    /+ ↓ output path: beside the pod's zip and digests, not under
         <lang>/abstraction/. the database carries its own images, so it is a
         distributable artefact in its own right and belongs with the others:
           pod/<doc>/            the source tree
           pod/<doc>.zip         the same, zipped
           pod/<doc>.digests.txt what is in them
           pod/<doc>.ocda.db     the abstraction, every language of it
    +/
    auto pths_pod = spinePathsPods!()(doc_matters);
    string base_pth = pths_pod.pod_dir_();
    try {
      if (!exists(base_pth)) {
        base_pth.mkdirRecurse;
      }
    } catch (Exception ex) {
    }
    /+ ↓ named for the document, not for one of its languages: the file holds
         every language, so the name is doc_uid_out without its ".<lang>".
    +/
    string db_file = ((base_pth.chainPath(
      doc_matters.src.doc_uid_out_no_lang ~ ".ocda.db")).asNormalizedPath).array;

    /+ ↓ the file is started fresh on the first language of the document and
         added to by the rest.
         .
         It used to be removed on every call, which was right when a call wrote
         the whole file. Now a call writes one language of it, so removing would
         leave only the last language of a ten language document (discarding the
         others). Which language is first comes from the manifest, not from the
         order the loop happens to run in, so this does not depend on the run.
    +/
    if (doc_matters.src.language == doc_matters.pod.manifest_list_of_languages[0]) {
      try {
        if (exists(db_file)) {
          remove(db_file);
        }
      } catch (Exception ex) {
      }
    }

    if (doc_matters.opt.action.vox_gt_1) {
      writeln(" ", db_file);
    }

    /+ ↓ open database and create schema +/
    auto db = Database(db_file);
    /+ ↓ write once, read many: no WAL.
         a WAL stamped file cannot be opened by a reader that cannot write
         beside it (SQLITE_READONLY on a read-only directory, and inside a
         zip or on read-only media), and this database has one writer and no
         concurrency, so WAL buys nothing here. MEMORY journal keeps the bulk
         load fast without leaving that mark on the file.
    +/
    db.run("PRAGMA journal_mode=MEMORY");
    db.run("PRAGMA synchronous=OFF");

    db.run("
      -- documents: a row per language of the document this database holds,
      -- and a database holds the document's whole set: ten rows for a ten
      -- language document, written one language at a time into the one file.
      -- Everything a reader needs to tell one language's rows from another's
      -- keys on documents.id, called doc_id wherever it is referred to.
      CREATE TABLE IF NOT EXISTS documents (
        id              INTEGER PRIMARY KEY,
        lang            TEXT NOT NULL,
        source_filename TEXT,
        ssp_digest      TEXT,
        UNIQUE(lang)
      );
      -- metadata: the document's own header, a row per property.
      -- Keyed on (doc_id, key) and not on key alone: every language has a
      -- title and a creator, and a key-only primary key refuses the second
      -- one. schema.name and schema.version describe the file rather than a
      -- document in it, so they are written with doc_id null and are the
      -- only rows that are.
      CREATE TABLE IF NOT EXISTS metadata (
        doc_id INTEGER REFERENCES documents(id),
        key    TEXT NOT NULL,
        value  TEXT NOT NULL,
        UNIQUE(doc_id, key)
      );
      -- the file level rows need a uniqueness rule of their own: in SQL no
      -- null equals any other null, so UNIQUE(doc_id, key) does not constrain
      -- them and schema.name would be written once per language. A partial
      -- index over key alone, for the null rows, is what makes them singular
      CREATE UNIQUE INDEX IF NOT EXISTS idx_metadata_file_key
        ON metadata(key) WHERE doc_id IS NULL;
      -- objects: one row per abstraction object, in document order.
      -- the fixed width arrays (ancestors, dom status, children, heading
      -- ancestors text, table widths and aligns) are JSON arrays so that
      -- json_extract, json_each and json_array_length can reach inside them;
      -- the open ended lists (images, links, anchor tags, subtoc) are rows
      -- in their own tables, not columns
      CREATE TABLE IF NOT EXISTS objects (
        id                    INTEGER PRIMARY KEY,
        doc_id                INTEGER NOT NULL REFERENCES documents(id),
        section               TEXT NOT NULL,
        seq                   INTEGER NOT NULL,
        ocn                   INTEGER DEFAULT 0,
        is_a                  TEXT NOT NULL,
        is_of_part            TEXT,
        is_of_section         TEXT,
        is_of_type            TEXT,
        heading_level         INTEGER,
        heading_lev_collapsed INTEGER,
        identifier            TEXT,
        parent_ocn            INTEGER DEFAULT 0,
        parent_lev            INTEGER DEFAULT 0,
        last_descendant_ocn   INTEGER DEFAULT 0,
        ocn_range_first       INTEGER DEFAULT 0,
        ocn_range_last        INTEGER DEFAULT 0,
        children              TEXT
          CHECK(children IS NULL OR json_valid(children)),
        ancestors             TEXT
          CHECK(ancestors IS NULL OR json_valid(ancestors)),
        ancestors_collapsed   TEXT
          CHECK(ancestors_collapsed IS NULL OR json_valid(ancestors_collapsed)),
        dom_status            TEXT
          CHECK(dom_status IS NULL OR json_valid(dom_status)),
        dom_status_collapsed  TEXT
          CHECK(dom_status_collapsed IS NULL OR json_valid(dom_status_collapsed)),
        dummy_heading         INTEGER DEFAULT 0,
        object_number_off     INTEGER DEFAULT 0,
        attrib                TEXT,
        meta_lang             TEXT,
        meta_syntax           TEXT,
        sha256                TEXT,
        indent_base           INTEGER DEFAULT 0,
        indent_hang           INTEGER DEFAULT 0,
        bullet                INTEGER DEFAULT 0,
        lang                  TEXT,
        has_links             INTEGER DEFAULT 0,
        has_notes_reg         INTEGER DEFAULT 0,
        has_notes_star        INTEGER DEFAULT 0,
        has_images            INTEGER DEFAULT 0,
        has_images_no_dim     INTEGER DEFAULT 0,
        segment               TEXT,
        segment_prev          TEXT,
        segment_next          TEXT,
        segment_epub          TEXT,
        segment_html_is       TEXT,
        segment_epub_is       TEXT,
        segment_lv4_is        TEXT,
        anchor                TEXT,
        heading_lev_anchor    TEXT,
        heading_ancestors_text TEXT
          CHECK(heading_ancestors_text IS NULL OR json_valid(heading_ancestors_text)),
        table_cols            INTEGER,
        table_widths          TEXT
          CHECK(table_widths IS NULL OR json_valid(table_widths)),
        table_aligns          TEXT
          CHECK(table_aligns IS NULL OR json_valid(table_aligns)),
        table_header          INTEGER,
        code_linenumbers      INTEGER DEFAULT 0,
        text                  TEXT,
        UNIQUE(doc_id, section, seq)
      );

      CREATE INDEX IF NOT EXISTS idx_objects_section ON objects(doc_id, section);
      CREATE INDEX IF NOT EXISTS idx_objects_ocn ON objects(ocn);
      CREATE INDEX IF NOT EXISTS idx_objects_parent ON objects(parent_ocn);
      CREATE INDEX IF NOT EXISTS idx_objects_is_a ON objects(is_a);
      CREATE INDEX IF NOT EXISTS idx_objects_heading ON objects(heading_level)
        WHERE heading_level IS NOT NULL;

      -- files: what this database carries with it, so that a reader needs
      -- nothing beside it. role says what a file is; images are the ones the
      -- output writers must have in order to produce a document. bytes are
      -- stored exactly as read, so sha256 matches the digest the abstraction
      -- was built with
      --
      --   image     what the output writers need to produce a document.
      --             named by its bare filename, as the abstraction refers
      --             to it, and shared by every language.
      --   source    the markup: the document file and the files it inserts,
      --             one set per language. named by its path within the pod,
      --             media/text/<lang>/<file>, because the bare names repeat
      --             across languages and because that is where a pod
      --             materialised from this database has to put them.
      --   conf      conf/document_make, the document's own configuration.
      --   manifest  pod.manifest, as the author wrote it.
      --   tools     tools/po4a, the translation catalogues, the same tree
      --             the pod carries: a pod written back out of this file is
      --             then the pod that went in, and not most of it.
      --
      -- source, conf and manifest are what make this a document source and
      -- not only a serialised abstraction: with them a pod can be written
      -- back out of the file and rebuilt from the markup, rather than only
      -- rendered from the objects.
      -- bytes and sha256 are always those of the ORIGINAL file, whatever
      -- compression says, so that a digest check, a digests.txt line and a
      -- source.digest rebuilt from these rows all keep working, and a reader
      -- that does not decompress can still say what it is looking at.
      --
      -- compression says how data is stored: NULL or 'none' for the bytes as
      -- read, 'zstd' for a zstd frame. It is decided by role and not by size,
      -- so the same kind of file always stores the same way and no reader has
      -- to measure anything to predict it: the text roles are compressed and
      -- images are not, png and jpeg being compressed already.
      CREATE TABLE IF NOT EXISTS files (
        id          INTEGER PRIMARY KEY,
        role        TEXT NOT NULL,
        name        TEXT NOT NULL,
        bytes       INTEGER NOT NULL,
        sha256      TEXT,
        width       INTEGER,
        height      INTEGER,
        compression TEXT,
        data        BLOB,
        UNIQUE(role, name)
      );

      -- the open ended per object lists, one row each, in order
      CREATE TABLE IF NOT EXISTS object_images (
        object_id INTEGER NOT NULL REFERENCES objects(id),
        seq       INTEGER NOT NULL,
        name      TEXT NOT NULL,
        bytes     INTEGER,
        sha256    TEXT,
        width     INTEGER,
        height    INTEGER,
        missing   INTEGER DEFAULT 0,
        PRIMARY KEY(object_id, seq)
      );
      CREATE TABLE IF NOT EXISTS object_links (
        object_id INTEGER NOT NULL REFERENCES objects(id),
        seq       INTEGER NOT NULL,
        url       TEXT NOT NULL,
        PRIMARY KEY(object_id, seq)
      );
      CREATE TABLE IF NOT EXISTS object_anchors (
        object_id INTEGER NOT NULL REFERENCES objects(id),
        seq       INTEGER NOT NULL,
        anchor    TEXT NOT NULL,
        PRIMARY KEY(object_id, seq)
      );
      CREATE TABLE IF NOT EXISTS object_subtoc (
        object_id INTEGER NOT NULL REFERENCES objects(id),
        seq       INTEGER NOT NULL,
        entry     TEXT NOT NULL,
        PRIMARY KEY(object_id, seq)
      );

      CREATE INDEX IF NOT EXISTS idx_files_role ON files(role);
      CREATE INDEX IF NOT EXISTS idx_object_images_name ON object_images(name);

      -- full text over the object text, external content so the text is not
      -- stored twice. per document: this is for exploring one abstraction,
      -- the collection wide search database is a different thing
      CREATE VIRTUAL TABLE IF NOT EXISTS objects_fts USING fts5(
        text, content='objects', content_rowid='id'
      );

      -- the shapes worth asking for, named
      -- both name the language, and order by it first. A view over a file
      -- that may hold several languages and does not say which would read
      -- as one document and be several: the language is in the projection
      -- so that what comes back says what it is, and a reader wanting one
      -- language adds WHERE lang = '..' rather than knowing to.
      CREATE VIEW IF NOT EXISTS outline AS
        SELECT d.lang AS lang, o.doc_id, o.section, o.seq, o.ocn,
               o.heading_level, o.heading_lev_collapsed,
               o.last_descendant_ocn, o.identifier, o.text
          FROM objects o JOIN documents d ON d.id = o.doc_id
         WHERE o.is_a = 'heading' ORDER BY o.doc_id, o.id;
      CREATE VIEW IF NOT EXISTS citable AS
        SELECT d.lang AS lang, o.*
          FROM objects o JOIN documents d ON d.id = o.doc_id
         WHERE o.ocn > 0 ORDER BY o.doc_id, o.ocn;
      CREATE VIEW IF NOT EXISTS document_files AS
        SELECT role, name, bytes, sha256, width, height FROM files ORDER BY role, name;
    ");

    /+ ↓ populate metadata, from the header blocks the .ssp gave back.
         one list, in ssp.d, not two: a property added to the .ssp header
         arrives here without anything being changed, and one that is not in
         the .ssp cannot be here at all. the keys are the .ssp's own
         (title.main, make.doc_type, doc_has.images), prefixed by their block
         where the block name is not already part of the key.
    +/
    db.run("BEGIN TRANSACTION");

    /+ ↓ the document this call is writing: one row in documents, and its id
         is the doc_id every row of this language then carries. Written with
         OR REPLACE on the language so that re-writing a language replaces
         its row rather than refusing it.
    +/
    long _doc_id;
    {
      auto doc_stmt = db.prepare(
        "INSERT OR REPLACE INTO documents (lang, source_filename, ssp_digest)"
        ~ " VALUES (:lang, :fn, :dig)"
      );
      doc_stmt.bind(":lang", doc_matters.src.language);
      doc_stmt.bind(":fn",   ssp_doc.source);
      doc_stmt.bind(":dig",  ssp_doc.ssp_digest);
      doc_stmt.execute();
      doc_stmt.finalize();
      _doc_id = db.lastInsertRowid;
    }

    auto meta_stmt = db.prepare(
      "INSERT INTO metadata (doc_id, key, value) VALUES (:doc_id, :key, :value)"
    );

    void insertMeta(string key, string value) {
      if (value.length > 0) {
        meta_stmt.bind(":doc_id", _doc_id);
        meta_stmt.bind(":key", key);
        meta_stmt.bind(":value", value);
        meta_stmt.execute();
        meta_stmt.reset();
      }
    }
    /+ ↓ what the file is rather than what a document in it is, so these two
         carry no doc_id and are the only rows that do not
    +/
    void insertFileMeta(string key, string value) {
      auto _s = db.prepare(
        "INSERT OR REPLACE INTO metadata (doc_id, key, value)"
        ~ " VALUES (NULL, :key, :value)"
      );
      _s.bind(":key", key);
      _s.bind(":value", value);
      _s.execute();
      _s.finalize();
    }
    void insertBlock(string prefix, string[] keys_in_order, string[string] kv) {
      foreach (k; keys_in_order) {
        if (k in kv) { insertMeta(prefix ~ k, kv[k]); }
      }
    }

    /+ ↓ what this file is, so a reader can tell before it starts. the
         version is the .ssp's own: the two serialisations are one format and
         move together
    +/
    insertFileMeta("schema.name", "sisu-abstraction-db");
    insertFileMeta("schema.version", ssp_format_version);
    /+ ↓ which spine wrote the file. Not a property of the document and not
         in the .ssp: it says what to blame when a database made months ago
         reads oddly, and it is the one thing a file cannot be asked for
         after the fact.
    +/
    insertFileMeta("build.spine_version", doc_matters.generator_program.ver);
    insertMeta("source.filename", ssp_doc.source);
    insertBlock("source.", ssp_doc.source_info_order, ssp_doc.source_info);
    /+ ↓ the digest of the .ssp this database was built from. a file cannot
         hold its own hash, so this one is not in the .ssp and is taken here:
         source.digest says which markup the abstraction came from,
         source.ssp_digest says which abstraction this database came from,
         and the chain .sst -> .ssp -> .ocda.db is checkable end to end
    +/
    insertMeta("source.ssp_digest", ssp_doc.ssp_digest);

    insertBlock("",         ssp_doc.meta_order,    ssp_doc.meta);
    insertBlock("make.",    ssp_doc.make_order,    ssp_doc.make);
    insertBlock("doc_has.", ssp_doc.doc_has_order, ssp_doc.doc_has);
    meta_stmt.finalize();

    /+ ↓ populate objects +/
    auto obj_stmt = db.prepare(
      "INSERT INTO objects ("
      ~ "doc_id, section, seq, ocn, is_a, is_of_part, is_of_section, is_of_type,"
      ~ "heading_level, heading_lev_collapsed, identifier,"
      ~ "parent_ocn, parent_lev, last_descendant_ocn, ocn_range_first, ocn_range_last,"
      ~ "children, ancestors, ancestors_collapsed,"
      ~ "dom_status, dom_status_collapsed,"
      ~ "dummy_heading, object_number_off,"
      ~ "attrib, meta_lang, meta_syntax, sha256,"
      ~ "indent_base, indent_hang, bullet, lang,"
      ~ "has_links, has_notes_reg, has_notes_star, has_images, has_images_no_dim,"
      ~ "segment, segment_prev, segment_next, segment_epub,"
      ~ "segment_html_is, segment_epub_is, segment_lv4_is,"
      ~ "anchor, heading_lev_anchor, heading_ancestors_text,"
      ~ "table_cols, table_widths, table_aligns, table_header,"
      ~ "code_linenumbers, text"
      ~ ") VALUES ("
      ~ ":doc_id, :section, :seq, :ocn, :is_a, :is_of_part, :is_of_section, :is_of_type,"
      ~ ":heading_level, :heading_lev_collapsed, :identifier,"
      ~ ":parent_ocn, :parent_lev, :last_descendant_ocn, :ocn_range_first, :ocn_range_last,"
      ~ ":children, :ancestors, :ancestors_collapsed,"
      ~ ":dom_status, :dom_status_collapsed,"
      ~ ":dummy_heading, :object_number_off,"
      ~ ":attrib, :meta_lang, :meta_syntax, :sha256,"
      ~ ":indent_base, :indent_hang, :bullet, :lang,"
      ~ ":has_links, :has_notes_reg, :has_notes_star, :has_images, :has_images_no_dim,"
      ~ ":segment, :segment_prev, :segment_next, :segment_epub,"
      ~ ":segment_html_is, :segment_epub_is, :segment_lv4_is,"
      ~ ":anchor, :heading_lev_anchor, :heading_ancestors_text,"
      ~ ":table_cols, :table_widths, :table_aligns, :table_header,"
      ~ ":code_linenumbers, :text"
      ~ ")"
    );

    /+ ↓ the per object lists, one row each +/
    auto img_stmt = db.prepare(
      "INSERT INTO object_images"
      ~ " (object_id, seq, name, bytes, sha256, width, height, missing)"
      ~ " VALUES (:object_id, :seq, :name, :bytes, :sha256, :width, :height, :missing)"
    );
    auto lnk_stmt = db.prepare(
      "INSERT INTO object_links (object_id, seq, url)"
      ~ " VALUES (:object_id, :seq, :url)"
    );
    auto anc_stmt = db.prepare(
      "INSERT INTO object_anchors (object_id, seq, anchor)"
      ~ " VALUES (:object_id, :seq, :anchor)"
    );
    auto sub_stmt = db.prepare(
      "INSERT INTO object_subtoc (object_id, seq, entry)"
      ~ " VALUES (:object_id, :seq, :entry)"
    );

    string[] section_order = ["head", "toc", "body", "endnotes",
      "glossary", "bibliography", "bookindex", "blurb", "tail"];

    foreach (section; section_order) {
      if (section !in doc_abstraction) continue;
      auto section_objs = doc_abstraction[section];
      if (section_objs.length == 0) continue;

      foreach (seq, obj; section_objs) {
        obj_stmt.bind(":doc_id", _doc_id);
        obj_stmt.bind(":section", section);
        obj_stmt.bind(":seq", cast(int) seq);
        obj_stmt.bind(":ocn", obj.metainfo.ocn);
        obj_stmt.bind(":is_a", obj.metainfo.is_a);

        /+ ↓ nullable string fields +/
        void bindStr(string param, string val) {
          import std.typecons : Nullable;
          if (val.length > 0) {
            obj_stmt.bind(param, val);
          } else {
            obj_stmt.bind(param, Nullable!string());
          }
        }
        /+ ↓ nullable int fields +/
        void bindInt(string param, int val, bool is_set) {
          import std.typecons : Nullable;
          if (is_set) {
            obj_stmt.bind(param, val);
          } else {
            obj_stmt.bind(param, Nullable!int());
          }
        }
        /+ ↓ fixed width int array as a JSON array, NULL when all zero +/
        void bindJsonInts(T)(string param, T arr) {
          import std.typecons : Nullable;
          bool any = false;
          foreach (a; arr) { if (a != 0) { any = true; break; } }
          if (any) {
            string[] _j;
            foreach (a; arr) { _j ~= a.to!string; }
            obj_stmt.bind(param, "[" ~ _j.join(",") ~ "]");
          } else {
            obj_stmt.bind(param, Nullable!string());
          }
        }
        /+ ↓ string array as a JSON array, NULL when every slot is empty +/
        void bindJsonStrs(T)(string param, T arr) {
          import std.json : JSONValue;
          import std.typecons : Nullable;
          bool any = false;
          foreach (s; arr) { if (s.length > 0) { any = true; break; } }
          if (any) {
            string[] _s;
            foreach (s; arr) { _s ~= s.to!string; }
            obj_stmt.bind(param, JSONValue(_s).toString);
          } else {
            obj_stmt.bind(param, Nullable!string());
          }
        }

        bindStr(":is_of_part", obj.metainfo.is_of_part);
        bindStr(":is_of_section", obj.metainfo.is_of_section);
        bindStr(":is_of_type", obj.metainfo.is_of_type);

        /+ ↓ heading levels, marked up and collapsed, 9 meaning unset +/
        bindInt(":heading_level", obj.metainfo.heading_lev_markup,
          (obj.metainfo.is_a == "heading" && obj.metainfo.heading_lev_markup < 9));
        bindInt(":heading_lev_collapsed", obj.metainfo.heading_lev_collapsed,
          (obj.metainfo.heading_lev_collapsed < 9));

        bindStr(":identifier", obj.metainfo.identifier);
        obj_stmt.bind(":parent_ocn", obj.metainfo.parent_ocn);
        obj_stmt.bind(":parent_lev", obj.metainfo.parent_lev_markup);
        obj_stmt.bind(":last_descendant_ocn", obj.metainfo.last_descendant_ocn);
        obj_stmt.bind(":ocn_range_first", obj.metainfo.ocn_range_first);
        obj_stmt.bind(":ocn_range_last", obj.metainfo.ocn_range_last);

        bindJsonInts(":children", obj.metainfo.children_headings);
        bindJsonInts(":ancestors", obj.metainfo.markedup_ancestors);
        bindJsonInts(":ancestors_collapsed", obj.metainfo.collapsed_ancestors);
        bindJsonInts(":dom_status", obj.metainfo.dom_structure_markedup_tags_status);
        bindJsonInts(":dom_status_collapsed", obj.metainfo.dom_structure_collapsed_tags_status);

        obj_stmt.bind(":dummy_heading", obj.metainfo.dummy_heading ? 1 : 0);
        obj_stmt.bind(":object_number_off", obj.metainfo.object_number_off ? 1 : 0);
        bindStr(":attrib", obj.metainfo.attrib);
        bindStr(":meta_lang", obj.metainfo.lang);
        bindStr(":meta_syntax", obj.metainfo.syntax);

        /+ ↓ object digest, hex, absent when never taken +/
        {
          import std.typecons : Nullable;
          bool has_sha = false;
          foreach (b; obj.metainfo.sha256.text) {
            if (b != 0) { has_sha = true; break; }
          }
          if (has_sha) {
            obj_stmt.bind(":sha256", obj.metainfo.sha256.text.toHexString.to!string);
          } else {
            obj_stmt.bind(":sha256", Nullable!string());
          }
        }

        // obj_stmt.bind(":dummy_heading", obj.metainfo.dummy_heading ? 1 : 0);
        // obj_stmt.bind(":object_number_off", obj.metainfo.object_number_off ? 1 : 0);
        obj_stmt.bind(":indent_base", obj.attrib.indent_base);
        obj_stmt.bind(":indent_hang", obj.attrib.indent_hang);
        obj_stmt.bind(":bullet", obj.attrib.bullet ? 1 : 0);
        bindStr(":lang", obj.attrib.language);
        obj_stmt.bind(":has_links", obj.has.inline_links ? 1 : 0);
        obj_stmt.bind(":has_notes_reg", obj.has.inline_notes_reg ? 1 : 0);
        obj_stmt.bind(":has_notes_star", obj.has.inline_notes_star ? 1 : 0);
        obj_stmt.bind(":has_images", obj.has.images ? 1 : 0);
        obj_stmt.bind(":has_images_no_dim", obj.has.image_without_dimensions ? 1 : 0);
        bindStr(":segment", obj.tags.in_segment_html);
        bindStr(":segment_prev", obj.tags.segname_prev);
        bindStr(":segment_next", obj.tags.segname_next);
        bindStr(":segment_epub", obj.tags.segment_anchor_tag_epub);
        bindStr(":segment_html_is", obj.tags.html_segment_anchor_tag_is);
        bindStr(":segment_epub_is", obj.tags.epub_segment_anchor_tag_is);
        bindStr(":segment_lv4_is", obj.tags.segment_lv4_is);
        bindStr(":anchor", obj.tags.anchor_tag_html);
        bindStr(":heading_lev_anchor", obj.tags.heading_lev_anchor_tag);
        bindJsonStrs(":heading_ancestors_text", obj.tags.heading_ancestors_text);

        /+ ↓ table properties +/
        {
          import std.typecons : Nullable;
          if (obj.metainfo.is_a == "table" && obj.table.number_of_columns > 0) {
            obj_stmt.bind(":table_cols", obj.table.number_of_columns);
            if (obj.table.column_widths.length > 0) {
              string[] ws;
              foreach (w; obj.table.column_widths) ws ~= w.to!string;
              obj_stmt.bind(":table_widths", "[" ~ ws.join(",") ~ "]");
            } else {
              obj_stmt.bind(":table_widths", Nullable!string());
            }
            if (obj.table.column_aligns.length > 0) {
              import std.json : JSONValue;
              string[] _al;
              foreach (a; obj.table.column_aligns) { _al ~= a.to!string; }
              obj_stmt.bind(":table_aligns", JSONValue(_al).toString);
            } else {
              obj_stmt.bind(":table_aligns", Nullable!string());
            }
            obj_stmt.bind(":table_header", obj.table.heading ? 1 : 0);
          } else {
            obj_stmt.bind(":table_cols", Nullable!int());
            obj_stmt.bind(":table_widths", Nullable!string());
            obj_stmt.bind(":table_aligns", Nullable!string());
            obj_stmt.bind(":table_header", Nullable!int());
          }
        }

        /+ ↓ code block properties +/
        obj_stmt.bind(":code_linenumbers",
          ((obj.metainfo.is_a == "code") && obj.code_block.linenumbers) ? 1 : 0);

        /+ ↓ text content +/
        bindStr(":text", obj.text);

        obj_stmt.execute();
        obj_stmt.reset();

        /+ ↓ the per object lists, keyed on the row just written +/
        long _obj_id = db.lastInsertRowid;
        {
          int _n = 0;
          foreach (i; obj.metainfo.sha256.images) {
            img_stmt.bind(":object_id", _obj_id);
            img_stmt.bind(":seq", _n);
            img_stmt.bind(":name", i.fileName);
            img_stmt.bind(":bytes", cast(long) i.fileSize);
            img_stmt.bind(":sha256", i.fileHash_sha256.toHexString.to!string);
            img_stmt.bind(":width", i.imageWidth);
            img_stmt.bind(":height", i.imageHeight);
            img_stmt.bind(":missing", i.fileMissing ? 1 : 0);
            img_stmt.execute();
            img_stmt.reset();
            ++_n;
          }
        }
        {
          int _n = 0;
          foreach (lnk; obj.stow.link) {
            if (lnk.length == 0) { continue; }
            lnk_stmt.bind(":object_id", _obj_id);
            lnk_stmt.bind(":seq", _n);
            lnk_stmt.bind(":url", lnk);
            lnk_stmt.execute();
            lnk_stmt.reset();
            ++_n;
          }
        }
        {
          int _n = 0;
          foreach (at; obj.tags.anchor_tags) {
            if (at.length == 0) { continue; }
            anc_stmt.bind(":object_id", _obj_id);
            anc_stmt.bind(":seq", _n);
            anc_stmt.bind(":anchor", at);
            anc_stmt.execute();
            anc_stmt.reset();
            ++_n;
          }
        }
        {
          int _n = 0;
          foreach (st; obj.tags.lev4_subtoc) {
            if (st.length == 0) { continue; }
            sub_stmt.bind(":object_id", _obj_id);
            sub_stmt.bind(":seq", _n);
            sub_stmt.bind(":entry", st);
            sub_stmt.execute();
            sub_stmt.reset();
            ++_n;
          }
        }
      }
    }

    obj_stmt.finalize();
    img_stmt.finalize();
    lnk_stmt.finalize();
    anc_stmt.finalize();
    sub_stmt.finalize();

    /+ ↓ the document's images, carried inside the file so that output can be
         produced from this database alone. bytes exactly as read, and the
         digest taken over them, so that it can be checked against the digest
         the abstraction recorded +/
    {
      import std.digest.sha : sha256Of;
      auto file_stmt = db.prepare(
        "INSERT OR IGNORE INTO files"
        ~ " (role, name, bytes, sha256, width, height, compression, data)"
        ~ " VALUES (:role, :name, :bytes, :sha256, :width, :height,"
        ~ " :compression, :data)"
      );
      int[string] _w, _h;
      foreach (section; section_order) {
        if (section !in doc_abstraction) { continue; }
        foreach (obj; doc_abstraction[section]) {
          foreach (i; obj.metainfo.sha256.images) {
            if (i.imageWidth > 0)  { _w[i.fileName] = i.imageWidth; }
            if (i.imageHeight > 0) { _h[i.fileName] = i.imageHeight; }
          }
        }
      }
      foreach (image; doc_matters.srcs.image_list) {
        string _fn = doc_matters.src.image_dir_path ~ "/" ~ image;
        ubyte[] _bytes;
        try {
          if (_fn.exists) { _bytes = cast(ubyte[]) _fn.read; }
        } catch (Exception ex) {
          _bytes = [];
        }
        if (_bytes.length == 0) {
          if (doc_matters.opt.action.vox_gt_1) {
            writeln("  image not carried into abstraction db: ", _fn);
          }
          continue;
        }
        file_stmt.bind(":role", "image");
        file_stmt.bind(":name", image);
        file_stmt.bind(":bytes", cast(long) _bytes.length);
        file_stmt.bind(":sha256", sha256Of(_bytes).toHexString.to!string);
        if (image in _w) { file_stmt.bind(":width", _w[image]); }
        else {
          import std.typecons : Nullable;
          file_stmt.bind(":width", Nullable!int());
        }
        if (image in _h) { file_stmt.bind(":height", _h[image]); }
        else {
          import std.typecons : Nullable;
          file_stmt.bind(":height", Nullable!int());
        }
        file_stmt.bind(":compression", "none");
        file_stmt.bind(":data", _bytes);
        file_stmt.execute();
        file_stmt.reset();
      }
      /+ ↓ the markup, the configuration and the manifest.
           Carried for the same reason the images are, and for one more: a
           database that holds the markup is a document source, and a pod can
           be written back out of it. The abstraction alone can be rendered
           but not re-parsed, and a reader who wants to correct a sentence
           needs the sentence as the author wrote it.
           .
           Stored as read, with the digest over those bytes, which is the
           digest the abstraction was built with: source.digest in the header
           is the sha256 over one line per markup file, and each line is
           checkable against the row written here.
      +/
      void _carry(string _role, string _name, string _path) {
        import std.typecons : Nullable;
        if (_path.length == 0 || _name.length == 0) { return; }
        ubyte[] _bytes;
        try {
          if (_path.exists && _path.isFile) { _bytes = cast(ubyte[]) _path.read; }
        } catch (Exception ex) {
          _bytes = [];
        }
        if (_bytes.length == 0) {
          if (doc_matters.opt.action.vox_gt_1) {
            writeln("  ", _role, " not carried into abstraction db: ", _path);
          }
          return;
        }
        /+ ↓ the text roles are stored compressed, the images are not.
             By role and not by size: the same kind of file then always
             stores the same way, and nobody has to measure a document to
             know how its database will look. The bytes and the digest
             recorded stay those of the original file either way.
             .
             A tiny file can come out a few bytes larger, a zstd frame
             having some overhead. Left alone: a rule to avoid that would
             be the size threshold this deliberately does not have.
        +/
        ubyte[] _stored      = _bytes;
        string  _compression = "none";
        try {
          _stored      = zstdCompress(_bytes, doc_matters.opt.action.pod_compression);
          _compression = "zstd";
        } catch (Exception ex) {
          stderr.writeln("WARNING: could not compress ", _name, ": ", ex.msg,
            "; stored as it stands");
          _stored      = _bytes;
          _compression = "none";
        }
        file_stmt.bind(":role",        _role);
        file_stmt.bind(":name",        _name);
        file_stmt.bind(":bytes",       cast(long) _bytes.length);
        file_stmt.bind(":sha256",      sha256Of(_bytes).toHexString.to!string);
        file_stmt.bind(":width",       Nullable!int());
        file_stmt.bind(":height",      Nullable!int());
        file_stmt.bind(":compression", _compression);
        file_stmt.bind(":data",        _stored);
        file_stmt.execute();
        file_stmt.reset();
      }
      {
        import std.path : baseName;
        /+ ↓ the markup of this language. The writer runs once per language,
             so the set of source rows fills up as the languages go by, and
             UNIQUE(role, name) over the pod relative path is what keeps a
             second run from writing them twice.
        +/
        string _text_dir = "media/text/" ~ doc_matters.src.lng ~ "/";
        _carry("source", _text_dir ~ doc_matters.src.filename,
          doc_matters.src.file_with_absolute_path.to!string);
        foreach (_insert; doc_matters.srcs.file_insert_list) {
          _carry("source", _text_dir ~ _insert.baseName, _insert);
        }
        /+ ↓ the configuration and the manifest, one of each for the whole
             document however many languages it has
        +/
        _carry("conf", "conf/document_make",
          doc_matters.src.conf_dir_path.to!string ~ "/document_make");
        if (doc_matters.src.is_pod) {
          _carry("manifest", "pod.manifest",
            doc_matters.pod.manifest_file_with_path);
        }
        /+ ↓ the translation catalogues, the same tools/po4a tree the pod
             writer carries and by the same rule: a directory walk, because a
             catalogue set is whatever the translator has, with every name
             checked before it is taken. Carrying less than the pod does
             would make a materialised pod a lossy copy of the original,
             which is the one thing this is for.
             .
             Done once, on the last language, as the pod writer does it (the
             tree is not language specific, and a pass per language would
             re-read all of it (5.5 MB ten times over, for live-manual) to
             insert rows the first pass already wrote).
        +/
        if (doc_matters.src.is_pod
          && doc_matters.src.language == doc_matters.pod.manifest_list_of_languages[$-1]
        ) {
          import sisudoc.ocda.io_in.carried_names;
          mixin spineCarriedNames;
          enum string _po4a_root = "tools/po4a";
          string _tools_in = doc_matters.pod.manifest_path ~ "/" ~ _po4a_root;
          if (exists(_tools_in) && _tools_in.isDir) {
            foreach (string _f; dirEntries(_tools_in, SpanMode.depth)) {
              if (!(_f.isFile)) { continue; }
              string _rel = _po4a_root ~ "/" ~ _f[(_tools_in.length + 1) .. $];
              string _bad = validateCarriedPath(_rel);
              if (_bad.length > 0) {
                writeln("WARNING not carried into the database: ", _bad);
                continue;
              }
              _carry("tools", _rel, _f);
            }
          }
        }
      }
      file_stmt.finalize();
    }

    /+ ↓ build the full text index from the rows just written +/
    /+ ↓ scoped to this document, not the whole table.
         (objects_fts is external content over objects.id, so re-running this
         for a second language would re-insert every row the first language
         put in: duplicate rowids, which fts5 does not refuse and which give
         duplicated and wrong results. Nothing in the test suite queries the
         index, so this one would not have been caught.)
    +/
    {
      auto _fts = db.prepare(
        "INSERT INTO objects_fts(rowid, text)"
        ~ " SELECT id, text FROM objects"
        ~ " WHERE doc_id = :doc_id AND text IS NOT NULL"
      );
      _fts.bind(":doc_id", _doc_id);
      _fts.execute();
      _fts.finalize();
    }

    db.run("COMMIT TRANSACTION");
  }
}
/+ ↓ a metadata row for one language of a database, written after the fact.
     .
     For the one thing a language cannot know while it is being written: how
     it compares with its siblings. The ocn alignment check can only run when
     every language of a document has been abstracted, which is after this
     database has been written, so the outcome is noted here in a pass of its
     own rather than held for by the writer.
     .
     A row that is about the artefact and not in the .ssp, as source.ssp_digest
     already is. It is INSERT OR REPLACE against UNIQUE(doc_id, key), so a
     second run over the same file restates rather than accumulates.
     .
     Its own template: writing one row should not instantiate the writer.
+/
template spineOcdaDbDocNote() {
  @trusted void spineOcdaDbDocNote(
    string db_file,
    string lang,
    string key,
    string value,
  ) {
    import std.file : exists;
    import d2sqlite3;
    if (db_file.length == 0 || !db_file.exists) { return; }
    try {
      auto db = Database(db_file);
      long _doc_id = -1;
      {
        auto _q = db.prepare("SELECT id FROM documents WHERE lang = :lang");
        _q.bind(":lang", lang);
        foreach (row; _q.execute()) { _doc_id = row["id"].as!long; }
        _q.finalize();
      }
      if (_doc_id < 0) { return; }
      auto _s = db.prepare(
        "INSERT OR REPLACE INTO metadata (doc_id, key, value)"
        ~ " VALUES (:doc_id, :key, :value)"
      );
      _s.bind(":doc_id", _doc_id);
      _s.bind(":key",    key);
      _s.bind(":value",  value);
      _s.execute();
      _s.finalize();
    } catch (Exception ex) {
      import std.stdio : stderr;
      stderr.writeln("WARNING: could not note ", key, " in ", db_file,
        ": ", ex.msg);
    }
  }
}
#+END_SRC

* org includes
** project version

#+NAME: spine_version
#+HEADER: :noweb yes
#+BEGIN_SRC emacs-lisp
<<./sisudoc_spine_version_info_and_doc_header_including_copyright_and_license.org:spine_project_version()>>
#+END_SRC

** year

#+NAME: year
#+HEADER: :noweb yes
#+BEGIN_SRC emacs-lisp
<<./sisudoc_spine_version_info_and_doc_header_including_copyright_and_license.org:year()>>
#+END_SRC

** document header including copyright & license

#+NAME: doc_header_including_copyright_and_license
#+HEADER: :noweb yes
#+BEGIN_SRC emacs-lisp
<<./sisudoc_spine_version_info_and_doc_header_including_copyright_and_license.org:spine_doc_header_including_copyright_and_license()>>
#+END_SRC

* __END__