kangshifu-board/docs/unified-database-migration-20261109/release-r2/package/components/research-runtime/scripts/unified_store.py

"""One durable store: the existing Parse PostgreSQL database.

Original rows live in typed PostgreSQL tables, not SQLite files or archive blobs.
An in-memory SQLite connection preserves the established FTS5/query contracts.
It is a disposable execution projection; there is no disk cache or fallback.
Monitoring writes hold a PostgreSQL advisory lock until commit/rollback and use
the same validated legacy rules. Research generations publish by atomic CAS.
"""
import base64
from contextlib import contextmanager
from contextvars import ContextVar
import hashlib
import json
import os
from pathlib import Path
import re
import sqlite3
import threading
import time

VERSION = 'ksf.unified-db.v1'
DATABASE = 'FKieRbIoGtmaster'
ROLE = 'FKieRbIoGtadmin'
CONTROL = 'ksf_voc_control'
_CACHE = {}
_LOCK = threading.RLock()
_BINDING = ContextVar('ksf_unified_binding', default=None)


class StoreError(ValueError):
    pass


def need(test, code):
    if not test:
        raise StoreError(code)


def canonical(value):
    return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(',', ':'), allow_nan=False)


def sha(value):
    return hashlib.sha256(value if isinstance(value, bytes) else value.encode('utf-8')).hexdigest()


def configured():
    return _BINDING.get() or os.environ.get('KSF_UNIFIED_DB_CONFIG')


@contextmanager
def binding(path=None):
    token = _BINDING.set(path or configured())
    try:
        yield
    finally:
        _BINDING.reset(token)


def configuration(path=None):
    path = Path(path or configured() or '').absolute()
    need(path.is_file() and not path.is_symlink(), 'UNIFIED_CONFIG_REQUIRED')
    if os.name != 'nt':
        need(path.stat().st_mode & 0o077 == 0, 'UNIFIED_CONFIG_NOT_PRIVATE')
    value = json.loads(path.read_text('utf-8-sig'))
    need(set(value) == {'schema', 'appId', 'database', 'connection'} and
         value['schema'] == VERSION and value['appId'] == 'FKieRbIoGt_main' and
         value['database'] == DATABASE, 'UNIFIED_TARGET_MISMATCH')
    conn = value['connection']
    allowed = {'service', 'host', 'port', 'user', 'passfile', 'sslmode', 'sslrootcert', 'connect_timeout'}
    need(isinstance(conn, dict) and conn and not set(conn) - allowed, 'INVALID_UNIFIED_CONNECTION')
    need('password' not in conn, 'USE_EXISTING_PRIVATE_PASSFILE')
    for key in ('passfile', 'sslrootcert'):
        if key in conn:
            p = Path(conn[key])
            need(p.is_absolute() and p.is_file() and not p.is_symlink(), 'UNIFIED_CONNECTION_FILE_REQUIRED')
            if key == 'passfile' and os.name != 'nt':
                need(p.stat().st_mode & 0o077 == 0, 'UNIFIED_PASSFILE_NOT_PRIVATE')
    return value


def connect_pg(path=None, readonly=False):
    pg = None
    try:
        import psycopg
        config = configuration(path)
        pg = psycopg.connect(dbname=DATABASE, application_name='ksf-unified-library',
                              **config['connection'])
        row = pg.execute('SELECT current_database(),current_user').fetchone()
        need(row == (DATABASE, ROLE), 'UNIFIED_TARGET_IDENTITY_MISMATCH')
        pg.commit()
        pg.execute('BEGIN ISOLATION LEVEL REPEATABLE READ READ ONLY' if readonly else 'BEGIN')
        pg.execute("SET LOCAL statement_timeout='120s'")
        pg.execute("SET LOCAL lock_timeout='20s'")
        return pg
    except StoreError:
        if pg is not None:
            pg.close()
        raise
    except Exception:
        if pg is not None:
            pg.close()
        raise StoreError('UNIFIED_CONNECTION_FAILED') from None


def identifier(value):
    need(isinstance(value, str) and re.fullmatch('[a-z_][a-z0-9_]{0,62}', value), 'INVALID_STORAGE_IDENTIFIER')
    return '"' + value + '"'


def ensure_control(pg):
    pg.execute('CREATE SCHEMA IF NOT EXISTS ' + CONTROL)
    pg.execute('''CREATE TABLE IF NOT EXISTS ksf_voc_control.datasets(
      dataset text PRIMARY KEY CHECK(dataset IN ('research','monitoring')),
      generation text NOT NULL, revision bigint NOT NULL, storage_schema text NOT NULL,
      manifest jsonb NOT NULL, pointer jsonb, updated_at timestamptz NOT NULL DEFAULT now())''')
    pg.execute('''CREATE TABLE IF NOT EXISTS ksf_voc_control.snapshots(
      dataset text NOT NULL, generation text NOT NULL, revision bigint NOT NULL,
      storage_schema text UNIQUE NOT NULL, manifest jsonb NOT NULL, pointer jsonb,
      created_at timestamptz NOT NULL DEFAULT now(), PRIMARY KEY(dataset,generation,revision))''')
    # Default grants are deliberately absent; existing Parse classes/ACLs are untouched.
    pg.execute('REVOKE ALL ON SCHEMA ksf_voc_control FROM PUBLIC')


def lock_dataset(pg, dataset):
    need(dataset in ('research', 'monitoring'), 'UNKNOWN_DATASET')
    pg.execute('SELECT pg_advisory_xact_lock(hashtext(%s))', ('ksf-unified:' + dataset,))


def descriptor(pg, dataset):
    row = pg.execute('SELECT generation,revision,storage_schema,manifest,pointer FROM '
                     'ksf_voc_control.datasets WHERE dataset=%s', (dataset,)).fetchone()
    need(row is not None, 'UNIFIED_DATASET_NOT_INSTALLED')
    return dict(zip(('generation', 'revision', 'storage_schema', 'manifest', 'pointer'), row))


def packed(value):
    return {'$bytes': base64.b64encode(value).decode('ascii')} if isinstance(value, bytes) else value


def row_hash(rows):
    digest = hashlib.sha256()
    for row in rows:
        digest.update(canonical([packed(v) for v in row]).encode('utf-8') + b'\n')
    return digest.hexdigest()


def inventory(c):
    """Store semantic tables and their original definitions; rebuild FTS internals."""
    objects = [dict(zip(('type', 'name', 'tbl_name', 'sql'), r)) for r in c.execute(
        "SELECT type,name,tbl_name,sql FROM sqlite_master WHERE sql IS NOT NULL ORDER BY type,name")]
    virtual = {r['name'] for r in objects if r['type'] == 'table' and
               re.match(r'CREATE\s+VIRTUAL\s+TABLE', r['sql'], re.I)}
    tables = []
    for obj in objects:
        name = obj['name']
        if obj['type'] != 'table' or name.startswith('sqlite_') or any(
                name == v or name.startswith(v + '_') for v in virtual):
            continue
        identifier(name)
        cols = [dict(zip(('cid', 'name', 'type', 'notnull', 'dflt_value', 'pk'), row))
                for row in c.execute('PRAGMA table_info(' + identifier(name) + ')')]
        need(cols and all(re.fullmatch('[a-z_][a-z0-9_]*', x['name']) for x in cols), 'INVALID_SOURCE_COLUMNS')
        pk = [x['name'] for x in sorted(cols, key=lambda x: x['pk']) if x['pk']]
        order = pk or [x['name'] for x in cols]
        has_rowid = not re.search(r'\bWITHOUT\s+ROWID\b', obj['sql'], re.I)
        need(not any(x['name'] in ('_rowid_', '_ksf_source_rowid') for x in cols), 'RESERVED_SOURCE_COLUMN')
        rows = list(c.execute('SELECT ' + ','.join(identifier(x['name']) for x in cols) +
                              ' FROM ' + identifier(name) + ' ORDER BY ' + ','.join(map(identifier, order))))
        indexes = []
        for r in c.execute('PRAGMA index_list(' + identifier(name) + ')'):
            fields = [x[2] for x in c.execute('PRAGMA index_info(' + identifier(r[1]) + ')')]
            if r[2] and all(fields):
                indexes.append(fields)
        tables.append({'name': name, 'columns': cols, 'primaryKey': pk, 'order': order,
                       'unique': indexes, 'count': len(rows), 'sha256': row_hash(rows),
                       'hasRowid': has_rowid, 'rowidSha256': row_hash(c.execute('SELECT _rowid_ FROM ' +
                       identifier(name) + ' ORDER BY ' + ','.join(map(identifier, order)))) if has_rowid else None})
    names = {t['name'] for t in tables}
    definitions = [o for o in objects if o['name'] in names or o['tbl_name'] in names or o['name'] in virtual]
    # Shadow tables cannot be recreated alongside their virtual table.
    definitions = [o for o in definitions if not any(o['name'].startswith(v + '_') for v in virtual)]
    result = {'schema': VERSION, 'tables': tables, 'definitions': definitions, 'virtualTables': sorted(virtual),
              'analyzed': bool(c.execute("SELECT 1 FROM sqlite_master WHERE name='sqlite_stat1'").fetchone())}
    result['fingerprint'] = sha(canonical(result))
    return result


def compare_manifest(before, after):
    need(before['fingerprint'] == after['fingerprint'], 'UNIFIED_FULL_ROW_VERIFICATION_FAILED')


def pg_type(field):
    kind = field['type'].upper()
    if 'INT' in kind:
        return 'bigint'
    if any(s in kind for s in ('REAL', 'FLOA', 'DOUB')):
        return 'double precision'
    if 'BLOB' in kind:
        return 'bytea'
    need(any(s in kind for s in ('TEXT', 'CHAR', 'CLOB')), 'UNSUPPORTED_SOURCE_TYPE')
    return 'text COLLATE "C"'


def upload_tables(pg, storage, manifest, c, previous=None):
    old = {t['name']: t for t in (previous or {}).get('tables', [])}
    need(not set(old) - {t['name'] for t in manifest['tables']}, 'MONITORING_TABLE_REMOVAL_REJECTED')
    for table in manifest['tables']:
        name = table['name']
        if name in old and old[name] == table:
            continue
        full = identifier(storage) + '.' + identifier(name)
        definitions = [identifier(f['name']) + ' ' + pg_type(f) + (' NOT NULL' if f['notnull'] else '')
                       for f in table['columns']]
        if table['hasRowid']:
            definitions.append('"_ksf_source_rowid" bigint NOT NULL UNIQUE')
        if table['primaryKey']:
            definitions.append('PRIMARY KEY (' + ','.join(map(identifier, table['primaryKey'])) + ')')
        if name in old:
            need(old[name]['columns'] == table['columns'], 'MONITORING_COLUMN_CHANGE_REJECTED')
            pg.execute('TRUNCATE TABLE ' + full)
        else:
            pg.execute('CREATE TABLE ' + full + '(' + ','.join(definitions) + ')')
        fields = ','.join(identifier(f['name']) for f in table['columns'])
        query = 'SELECT ' + fields + (',_rowid_' if table['hasRowid'] else '') + ' FROM ' + identifier(name) + ' ORDER BY ' + ','.join(map(identifier, table['order']))
        with pg.cursor().copy('COPY ' + full + '(' + fields + (',"_ksf_source_rowid"' if table['hasRowid'] else '') + ') FROM STDIN') as copy:
            for row in c.execute(query):
                copy.write_row(tuple(row))
        for idx, columns in enumerate(table['unique']):
            if name not in old and columns != table['primaryKey']:
                pg.execute('CREATE UNIQUE INDEX ON ' + full + '(' + ','.join(map(identifier, columns)) + ')')
        # Independent PG read of EVERY field and row, not just a count.
        rows = pg.execute('SELECT ' + fields + ' FROM ' + full + ' ORDER BY ' +
                          ','.join(map(identifier, table['order']))).fetchall()
        need(len(rows) == table['count'] and row_hash(rows) == table['sha256'], 'POSTGRES_ROW_HASH_MISMATCH')
        if table['hasRowid']:
            need(row_hash(pg.execute('SELECT "_ksf_source_rowid" FROM ' + full + ' ORDER BY ' +
                          ','.join(map(identifier, table['order'])))) == table['rowidSha256'], 'POSTGRES_ROWID_HASH_MISMATCH')


def stage_snapshot(pg, dataset, generation, revision, manifest, c, pointer=None):
    """Add a sealed generation in the SAME PostgreSQL database, never CREATE DATABASE."""
    storage = 'ksfv_' + sha(dataset + ':' + generation + ':' + str(revision) + ':' + manifest['fingerprint'])[:40]
    need(not pg.execute('SELECT 1 FROM pg_namespace WHERE nspname=%s', (storage,)).fetchone(),
         'SNAPSHOT_SCHEMA_ALREADY_EXISTS')
    pg.execute('CREATE SCHEMA ' + identifier(storage))
    pg.execute('REVOKE ALL ON SCHEMA ' + identifier(storage) + ' FROM PUBLIC')
    upload_tables(pg, storage, manifest, c)
    pg.execute('INSERT INTO ksf_voc_control.snapshots(dataset,generation,revision,storage_schema,manifest,pointer) '
               'VALUES(%s,%s,%s,%s,%s::jsonb,%s::jsonb)',
               (dataset, generation, revision, storage, canonical(manifest), canonical(pointer)))
    return storage


def reconstruct(pg, desc):
    """Ephemeral RAM projection; never connects to a SQLite filename."""
    manifest = desc['manifest']
    c = sqlite3.connect(':memory:', check_same_thread=False)
    try:
        c.execute('PRAGMA foreign_keys=OFF')
        definitions = manifest['definitions']
        virtual = manifest['virtualTables']
        for obj in definitions:
            if obj['type'] == 'table' and obj['name'] not in virtual:
                c.execute(obj['sql'])
        for table in manifest['tables']:
            fields = ','.join(identifier(f['name']) for f in table['columns'])
            extra = ',"_ksf_source_rowid"' if table['hasRowid'] else ''
            sql = 'SELECT ' + fields + extra + ' FROM ' + identifier(desc['storage_schema']) + '.' + identifier(table['name'])
            rows = pg.execute(sql + ' ORDER BY ' + ','.join(map(identifier, table['order']))).fetchall()
            original = [r[:-1] for r in rows] if table['hasRowid'] else rows
            need(len(rows) == table['count'] and row_hash(original) == table['sha256'], 'UNIFIED_ROW_CONTENT_CHANGED')
            if table['hasRowid']:
                need(row_hash((r[-1],) for r in rows) == table['rowidSha256'], 'UNIFIED_ROWID_CHANGED')
            c.executemany('INSERT INTO ' + identifier(table['name']) + '(' + fields +
                          (',_rowid_' if table['hasRowid'] else '') + ') VALUES(' +
                          ','.join('?' for _ in range(len(table['columns']) + int(table['hasRowid']))) + ')', rows)
        for obj in definitions:
            if obj['type'] in ('index', 'view', 'trigger') or obj['name'] in virtual:
                c.execute(obj['sql'])
        for name in virtual:
            need(name == 'retrieval_fts', 'UNSUPPORTED_VIRTUAL_TABLE')
            c.execute('INSERT INTO retrieval_fts(retrieval_fts) VALUES(\'rebuild\')')
        if manifest.get('analyzed'):
            c.execute('ANALYZE')
        c.commit()
        need(c.execute('PRAGMA integrity_check').fetchone()[0] == 'ok' and
             not c.execute('PRAGMA foreign_key_check').fetchall(), 'UNIFIED_PROJECTION_INTEGRITY_FAILED')
        compare_manifest(manifest, inventory(c))
        return c
    except BaseException:
        c.close()
        raise


def pointer(path=None, dataset='research'):
    with connect_pg(path, readonly=True) as pg:
        desc = descriptor(pg, dataset)
        return desc['pointer'] or {'generation': desc['generation']}


def research_reference(path=None):
    return Reference(path or configured())


class Reference:
    def __init__(self, path):
        need(bool(path), 'UNIFIED_CONFIG_REQUIRED')
        self.path = str(Path(path).absolute())


@contextmanager
def read_research(ref):
    # Requests are serial within a worker; PostgreSQL readers still use snapshots.
    with _LOCK:
        with connect_pg(ref.path, readonly=True) as pg:
            desc = descriptor(pg, 'research')
            key = (ref.path, desc['generation'], desc['revision'])
            if key not in _CACHE:
                fresh = reconstruct(pg, desc)
                for old in list(_CACHE):
                    if old[0] == ref.path:
                        _CACHE.pop(old).close()
                fresh.row_factory = sqlite3.Row
                fresh.execute('PRAGMA query_only=ON')
                _CACHE[key] = fresh
        c = _CACHE[key]
        c.execute('BEGIN')
        try:
            yield c
        finally:
            c.rollback()


def migrate(source, config, dataset='research', registry_pointer=None, apply=False, expected=None):
    source = Path(source).resolve()
    need(source.is_file(), 'MIGRATION_SOURCE_MISSING')
    # Source snapshot is read transactionally; WAL is included by SQLite backup, not file copy.
    raw = sqlite3.connect(source.as_uri() + '?mode=ro', uri=True)
    c = sqlite3.connect(':memory:')
    try:
        raw.backup(c)
        raw.close()
        need(c.execute('PRAGMA integrity_check').fetchone()[0] == 'ok' and
             not c.execute('PRAGMA foreign_key_check').fetchall(), 'MIGRATION_SOURCE_INTEGRITY_FAILED')
        manifest = inventory(c)
        meta = dict(c.execute('SELECT key,value FROM meta'))
        generation = meta.get('generation') if dataset == 'research' else manifest['fingerprint']
        need(bool(generation), 'SOURCE_GENERATION_MISSING')
        if dataset == 'research':
            need(meta.get('status') == 'ready' and meta.get('schema') == 'ksf.voc-library.v2', 'SOURCE_NOT_READY')
        if registry_pointer:
            need(registry_pointer.get('generation') == generation, 'SOURCE_POINTER_GENERATION_MISMATCH')
        result = {'ok': True, 'schema': VERSION, 'dataset': dataset, 'generation': generation,
                  'fingerprint': manifest['fingerprint'], 'tables': [{k: t[k] for k in ('name', 'count', 'sha256')}
                  for t in manifest['tables']], 'readOnly': not apply, 'database': DATABASE,
                  'applied': False, 'sourceModified': False, 'businessRowsChanged': 0}
        with connect_pg(config, readonly=not apply) as pg:
            control_exists = pg.execute('SELECT 1 FROM pg_namespace WHERE nspname=%s', (CONTROL,)).fetchone()
            prior = None
            if control_exists:
                row = pg.execute('SELECT generation,manifest FROM ksf_voc_control.datasets WHERE dataset=%s', (dataset,)).fetchone()
                if row:
                    prior = {'generation': row[0], 'manifest': row[1]}
            if prior and prior['generation'] == generation:
                compare_manifest(prior['manifest'], manifest)
                result['operation'] = 'already-migrated'
                # Do not claim a successful retry from metadata alone.
                check = reconstruct(pg, descriptor(pg, dataset))
                check.close()
                return result
            need(expected is None or (prior or {}).get('generation') == expected, 'UNIFIED_GENERATION_CHANGED')
            need(not prior or expected is not None, 'EXISTING_DATASET_REQUIRES_EXPECTED_GENERATION')
            if not apply:
                result['operation'] = 'plan'
                return result
            ensure_control(pg)
            lock_dataset(pg, dataset)
            current = pg.execute('SELECT generation FROM ksf_voc_control.datasets WHERE dataset=%s', (dataset,)).fetchone()
            need((current[0] if current else None) == (prior or {}).get('generation'), 'UNIFIED_CONCURRENT_CHANGE')
            revision = 1
            storage = stage_snapshot(pg, dataset, generation, revision, manifest, c, registry_pointer)
            pg.execute('INSERT INTO ksf_voc_control.datasets(dataset,generation,revision,storage_schema,manifest,pointer) '
                       'VALUES(%s,%s,%s,%s,%s::jsonb,%s::jsonb) ON CONFLICT(dataset) DO UPDATE SET '
                       'generation=EXCLUDED.generation,revision=EXCLUDED.revision,storage_schema=EXCLUDED.storage_schema,'
                       'manifest=EXCLUDED.manifest,pointer=EXCLUDED.pointer,updated_at=now()',
                       (dataset, generation, revision, storage, canonical(manifest), canonical(registry_pointer)))
            result.update(applied=True, operation='migrated', allRowsVerified=True)
        return result
    finally:
        raw.close()
        c.close()


def archive_research(source, config, entry, apply=False):
    """Retain an existing legacy rollback target inside PostgreSQL without activating it."""
    source = Path(source).resolve()
    need(source.is_file() and sha(source.read_bytes()) == entry.get('databaseSha256'), 'HISTORICAL_SOURCE_HASH_MISMATCH')
    raw = sqlite3.connect(source.as_uri() + '?mode=ro', uri=True)
    c = sqlite3.connect(':memory:')
    try:
        raw.backup(c)
        meta = dict(c.execute('SELECT key,value FROM meta'))
        need(meta.get('generation') == entry.get('generation') and meta.get('status') == 'ready', 'HISTORICAL_GENERATION_MISMATCH')
        need(c.execute('PRAGMA integrity_check').fetchone()[0] == 'ok' and
             not c.execute('PRAGMA foreign_key_check').fetchall(), 'HISTORICAL_SOURCE_INTEGRITY_FAILED')
        manifest = inventory(c)
        result = {'generation': meta['generation'], 'fingerprint': manifest['fingerprint'],
                  'tables': len(manifest['tables']), 'rows': sum(t['count'] for t in manifest['tables']),
                  'readOnly': not apply, 'activated': False}
        with connect_pg(config, readonly=not apply) as pg:
            exists = pg.execute('SELECT 1 FROM pg_namespace WHERE nspname=%s', (CONTROL,)).fetchone()
            if not apply:
                return result
            need(exists, 'MIGRATE_CURRENT_GENERATION_FIRST')
            lock_dataset(pg, 'research')
            row = pg.execute('SELECT storage_schema,manifest FROM ksf_voc_control.snapshots '
                             'WHERE dataset=\'research\' AND generation=%s AND revision=1', (meta['generation'],)).fetchone()
            if row:
                compare_manifest(row[1], manifest)
                verified = reconstruct(pg, {'storage_schema': row[0], 'manifest': row[1]});verified.close()
            else:
                saved_pointer = {**entry, 'history': entry.get('history', [entry])}
                stage_snapshot(pg, 'research', meta['generation'], 1, manifest, c, saved_pointer)
            result['allRowsVerified'] = True
        return result
    finally:
        raw.close();c.close()


class MonitoringConnection:
    """Legacy write transaction, serialized before reading its PostgreSQL baseline."""
    def __init__(self, config=None, write=True):
        self.config = config or configured()
        self.pg = connect_pg(self.config, readonly=not write)
        self.write = write
        self.closed = False
        try:
            if write:
                lock_dataset(self.pg, 'monitoring')
            self.desc = descriptor(self.pg, 'monitoring')
            self.c = reconstruct(self.pg, self.desc)
            if not write:
                self.c.execute('PRAGMA query_only=ON')
        except BaseException:
            self.pg.close()
            raise

    @property
    def row_factory(self):
        return self.c.row_factory

    @row_factory.setter
    def row_factory(self, value):
        self.c.row_factory = value

    def __getattr__(self, key):
        return getattr(self.c, key)

    def commit(self):
        need(self.write and not self.closed, 'UNIFIED_READ_ONLY')
        self.c.commit()
        need(not self.c.execute('PRAGMA foreign_key_check').fetchall(), 'MONITORING_FOREIGN_KEY_FAILED')
        manifest = inventory(self.c)
        if manifest['fingerprint'] != self.desc['manifest']['fingerprint']:
            revision = self.desc['revision'] + 1
            storage = 'ksfv_monitoring_live'
            exists = self.pg.execute('SELECT 1 FROM pg_namespace WHERE nspname=%s', (storage,)).fetchone()
            if not exists:
                self.pg.execute('CREATE SCHEMA ' + identifier(storage))
                self.pg.execute('REVOKE ALL ON SCHEMA ' + identifier(storage) + ' FROM PUBLIC')
            upload_tables(self.pg, storage, manifest, self.c,
                          self.desc['manifest'] if self.desc['storage_schema'] == storage else None)
            self.pg.execute('UPDATE ksf_voc_control.datasets SET revision=%s,storage_schema=%s,manifest=%s::jsonb,'
                            'updated_at=now() WHERE dataset=\'monitoring\' AND revision=%s',
                            (revision, storage, canonical(manifest), self.desc['revision']))
        self.pg.commit()
        self.closed = True

    def rollback(self):
        self.c.rollback()
        self.pg.rollback()
        self.closed = True

    def close(self):
        self.c.close()
        self.pg.close()


def monitoring_available(root):
    if configured():
        with connect_pg(readonly=True) as pg:
            descriptor(pg, 'monitoring')
        return True
    return bool(root) and (Path(root) / 'monitoring.sqlite').is_file()


def monitoring_connection(root, write=True, timeout=20):
    if configured():
        return MonitoringConnection(write=write)
    p = Path(root).resolve() / 'monitoring.sqlite'
    return sqlite3.connect(str(p) if write else p.as_uri() + '?mode=ro', uri=not write, timeout=timeout)


def publish(incoming, registry, apply=False, label='资料更新', rollback=None, expected_generation=None, review_correction=None):
    """Existing merge rules build only a disposable stage; publish in PostgreSQL."""
    import tempfile
    import update_library
    config = configured()
    need(config, 'UNIFIED_CONFIG_REQUIRED')
    if rollback:
        need(review_correction is None, 'CORRECTION_ROLLBACK_NOT_ALLOWED')
        return rollback_research(config, rollback, expected_generation, apply)
    with tempfile.TemporaryDirectory(prefix='ksf-unified-merge-') as td:
        baseline, target = Path(td) / 'baseline.sqlite', Path(td) / 'merged.sqlite'
        with connect_pg(config, readonly=True) as pg:
            desc = descriptor(pg, 'research')
            need(expected_generation is not None and desc['generation'] == expected_generation, 'LIBRARY_GENERATION_CHANGED')
            prior = desc['pointer'] or {'generation': desc['generation'], 'history': []}
            meta = update_library.verified(Path(incoming))
            if any(x.get('incomingGeneration') == meta['generation'] for x in prior.get('history', [])):
                return {'ok': True, 'applied': False, 'readOnly': not apply, 'operation': 'already-imported',
                        'generation': desc['generation'], 'currentGeneration': desc['generation'],
                        'plannedGeneration': desc['generation'], 'incomingGeneration': meta['generation'],
                        'plannedCounts': {t['name']: t['count'] for t in desc['manifest']['tables']},
                        'temporaryFilesUsed': False, 'persistentFilesWritten': False,
                        'activeLibraryModified': False, 'databaseWrites': 0}
            c = reconstruct(pg, desc)
            disk = sqlite3.connect(baseline)
            c.backup(disk)
            disk.close()
            c.close()
        delta = update_library.merge(baseline, Path(incoming), target, review_correction)
        final = update_library.verified(target)
        entry = {'generation': final['generation'], 'incomingGeneration': meta['generation'],
                 'delta': delta, 'label': label, 'updatedAt': __import__('datetime').datetime.now(
                     __import__('datetime').timezone.utc).isoformat(), 'storage': 'parse-postgresql'}
        pointer_value = {**entry, 'history': [entry, *prior.get('history', [])]}
        result = migrate(target, config, registry_pointer=pointer_value, apply=apply, expected=expected_generation)
        with update_library.voc_db.readonly(target) as c:
            counts = update_library.voc_db.table_counts(c)
            rows = [dict(r) for r in c.execute('SELECT entity,record_id,decision,reason_codes_json,prior_source_record_id,incoming_source_record_id,selected_source_record_id FROM merge_decisions WHERE batch_sha256=? ORDER BY entity,record_id LIMIT 101', (meta['archive_sha256'],))]
        for row in rows:
            row['reasons'] = json.loads(row.pop('reason_codes_json'))
        return {**result, **(pointer_value if apply else {}), 'delta': delta,
                'incomingGeneration': meta['generation'], 'currentGeneration': expected_generation,
                'plannedGeneration': final['generation'], 'plannedCounts': counts,
                'recordDecisions': rows[:100], 'recordDecisionsTruncated': len(rows) > 100,
                'databaseWrites': 'postgresql-new-generation' if apply else 'temporary-only',
                'activeLibraryModified': bool(result['applied']), 'temporaryFilesUsed': True,
                'persistentFilesWritten': bool(result['applied']), 'persistentSqliteFilesWritten': False}


def rollback_research(config, generation, expected, apply=False):
    """Switch between verified PostgreSQL research generations; monitoring is unchanged."""
    need(expected is not None, 'EXPECTED_GENERATION_REQUIRED')
    with connect_pg(config, readonly=not apply) as pg:
        if apply:
            lock_dataset(pg, 'research')
        current = descriptor(pg, 'research')
        need(current['generation'] == expected, 'UNIFIED_GENERATION_CHANGED')
        row = pg.execute('SELECT generation,revision,storage_schema,manifest,pointer FROM '
                         'ksf_voc_control.snapshots WHERE dataset=\'research\' AND generation=%s '
                         'ORDER BY revision DESC LIMIT 1', (generation,)).fetchone()
        need(row is not None, 'UNIFIED_VERSION_NOT_FOUND')
        target = dict(zip(('generation', 'revision', 'storage_schema', 'manifest', 'pointer'), row))
        c = reconstruct(pg, target)
        c.close()
        if apply:
            pg.execute('UPDATE ksf_voc_control.datasets SET generation=%s,revision=%s,storage_schema=%s,'
                       'manifest=%s::jsonb,pointer=%s::jsonb,updated_at=now() WHERE dataset=\'research\'',
                       (target['generation'], target['revision'], target['storage_schema'],
                        canonical(target['manifest']), canonical(target['pointer'])))
        return {'ok': True, 'readOnly': not apply, 'applied': bool(apply),
                'generation': generation if apply else expected, 'rollbackGeneration': generation,
                'currentGeneration': expected, 'databaseWrites': int(apply),
                'storage': 'parse-postgresql', 'monitoringChanged': False, 'sourceDeleted': False}