Build historical feature snapshots

"""Retain feature vintages; join only values published by the prediction time.

Synthetic UTC minutes and supplied publication times model a trusted writer.
This SQLite exercise is not a production event-time or publication service.
"""
import hashlib
import json
import sqlite3
from datetime import datetime, timezone
from pathlib import Path
from vendor.availability import instant, validate, check_id
from vendor.feature_math import Feature, compute

EPOCH = datetime(2026, 1, 1, tzinfo=timezone.utc)
COUNT = Feature('user_events_1d', 1, 'user', 1, 'count')


def minute(value):
    seconds = (instant(value) - EPOCH).total_seconds()
    if seconds < 0 or seconds % 60:
        raise ValueError('fixture timestamps must be whole minutes since 2026-01-01 UTC')
    return int(seconds // 60)


def nonnegative(value):
    if type(value) is not int or not 0 <= value < 2**62:
        raise ValueError('time and age must be bounded nonnegative integer minutes')
    return value


def source_rows(events):
    validate([], events, [])
    return [{'event': index, 'user': e['entity'], 'item': 'unused',
             'event_minute': minute(e['event_at']), 'ingested_minute': minute(e['arrived_at'])}
            for index, e in enumerate(events)]


class History:
    def __init__(self, path=':memory:'):
        self.db = sqlite3.connect(path)
        self.db.row_factory = sqlite3.Row
        self.db.executescript('''
            CREATE TABLE IF NOT EXISTS definitions (
                feature TEXT PRIMARY KEY, spec TEXT NOT NULL);
            CREATE TABLE IF NOT EXISTS batches (
                feature TEXT, cutoff INTEGER, published_at INTEGER, mode TEXT NOT NULL,
                source_sha256 TEXT NOT NULL,
                PRIMARY KEY(feature, cutoff, published_at));
            CREATE TABLE IF NOT EXISTS history (
                feature TEXT, entity TEXT, cutoff INTEGER, published_at INTEGER,
                value INTEGER NOT NULL, PRIMARY KEY(feature, entity, cutoff, published_at));
            CREATE TABLE IF NOT EXISTS online (
                feature TEXT, entity TEXT, cutoff INTEGER NOT NULL, published_at INTEGER NOT NULL,
                value INTEGER NOT NULL, PRIMARY KEY(feature, entity));
        ''')

    def close(self):
        self.db.close()

    def materialize(self, events, cutoff, published_at, mode='as_known', feature=COUNT):
        cutoff = nonnegative(cutoff); published_at = nonnegative(published_at)
        if published_at < cutoff:
            raise ValueError('publication cannot precede its feature cutoff')
        if mode not in ('as_known', 'corrected'):
            raise ValueError('unknown availability mode')
        if (not isinstance(feature, Feature) or feature.entity != 'user'
                or type(feature.version) is not int or feature.version < 1
                or type(feature.window_days) is not int or feature.window_days < 1
                or feature.aggregation not in ('count', 'distinct_days')):
            raise ValueError('expected a versioned user-count or active-day feature')
        check_id(feature.name)
        # A corrected backfill still cannot use an event unavailable at publication.
        available = [e for e in source_rows(events) if e['ingested_minute'] <= published_at]
        values = compute(feature, available, cutoff, mode)
        source_hash = hashlib.sha256(json.dumps(sorted(events, key=lambda e: e['id']),
                                                sort_keys=True).encode()).hexdigest()
        key = f'{feature.name}@v{feature.version}'
        with self.db:
            old = self.db.execute('SELECT spec FROM definitions WHERE feature=?', (key,)).fetchone()
            if old and old['spec'] != feature.spec():
                raise ValueError('changed definition requires a new version')
            identity = (key, cutoff, published_at)
            batch = self.db.execute('SELECT mode,source_sha256 FROM batches WHERE feature=? '
                                    'AND cutoff=? AND published_at=?', identity).fetchone()
            if batch:
                if tuple(batch) != (mode, source_hash):
                    raise ValueError('conflicting publication identity')
                return key  # Identical retry; never rewrite a published vintage.
            latest = self.db.execute('SELECT MAX(published_at) FROM batches WHERE feature=?', (key,)).fetchone()[0]
            if latest is not None and published_at < latest:
                raise ValueError('cannot backdate publication; use the actual later availability time')
            self.db.execute('INSERT OR IGNORE INTO definitions VALUES (?,?)', (key, feature.spec()))
            self.db.execute('INSERT INTO batches VALUES (?,?,?,?,?)', (*identity, mode, source_hash))
            for entity, value in values.items():
                self.db.execute('INSERT INTO history VALUES (?,?,?,?,?)', (key, entity, cutoff, published_at, value))
                self.db.execute('''INSERT INTO online VALUES (?,?,?,?,?)
                    ON CONFLICT(feature,entity) DO UPDATE SET
                        cutoff=excluded.cutoff, published_at=excluded.published_at, value=excluded.value
                    WHERE excluded.cutoff > online.cutoff OR
                        (excluded.cutoff=online.cutoff AND excluded.published_at>online.published_at)''',
                    (key, entity, cutoff, published_at, value))
        return key

    def point_in_time(self, feature, entity, predict_at):
        nonnegative(predict_at); check_id(feature); check_id(entity)
        query = (Path(__file__).resolve().parent / 'join.sql').read_text()
        row = self.db.execute(query, (feature, entity, predict_at, predict_at)).fetchone()
        return dict(row) if row else None

    def join(self, observations, feature):
        validate(observations, [], [])
        # Preserve one row per observation, even when the feature is missing.
        return [dict(row, feature=self.point_in_time(feature, row['entity'], minute(row['predict_at'])))
                for row in observations]

    def read_online(self, feature, entity, now, max_age):
        nonnegative(now); nonnegative(max_age); check_id(feature); check_id(entity)
        row = self.db.execute('SELECT value,cutoff,published_at FROM online WHERE feature=? AND entity=?',
                              (feature, entity)).fetchone()
        if row is None:
            return {'status': 'missing', 'value': None, 'age_minutes': None}
        age = now - row['cutoff']
        if row['cutoff'] > now or row['published_at'] > now:
            status = 'unavailable'
        else:
            status = 'fresh' if age <= max_age else 'stale'
        return dict(row, status=status, value=row['value'] if status == 'fresh' else None, age_minutes=age)

"""Inspect real local SQLite joins over an authored synthetic event log."""
import argparse
import hashlib
import json
from pathlib import Path
from history import History, minute

ROOT = Path(__file__).resolve().parent
KEY = 'user_events_1d@v1'
CUTOFF = minute('2026-01-08T12:00:00Z')
FIRST = minute('2026-01-08T12:30:00Z')
CORRECTION = minute('2026-01-08T14:00:00Z')


def events():
    return json.loads((ROOT / 'scenario.json').read_text())['events']


def observations():
    return [dict(id=id, entity=entity, predict_at='2026-01-08T'+time+'Z')
            for id, entity, time in [('before-job','user-01','12:15:00'),
                                    ('first-read','user-01','13:00:00'),
                                    ('after-correction','user-01','15:00:00'),
                                    ('unknown-user','user-99','15:00:00')]]


def record():
    db = History()
    try:
        db.materialize(events(), CUTOFF, FIRST)
        before = db.join(observations(), KEY)
        db.materialize(events(), CUTOFF, CORRECTION, 'corrected')
        after = db.join(observations(), KEY)
        return {'mode': 'local SQLite; authored synthetic data; no model',
                'input_sha256': hashlib.sha256((ROOT/'scenario.json').read_bytes()).hexdigest(),
                'cutoff': CUTOFF, 'first_published_at': FIRST, 'corrected_published_at': CORRECTION,
                'before_backfill': before, 'after_backfill': after,
                'historical_read_unchanged': before[1] == after[1],
                'retained_vintages_user_01': db.db.execute('SELECT COUNT(*) FROM history '
                                                        'WHERE entity=?', ('user-01',)).fetchone()[0]}
    finally:
        db.close()


def main():
    p = argparse.ArgumentParser(); p.add_argument('--check', action='store_true'); p.add_argument('--save', type=Path)
    args=p.parse_args(); result=record()
    if args.check and result != json.loads((ROOT/'join-result.json').read_text()):
        raise SystemExit('joined rows differ from the bundled record')
    if args.save: args.save.write_text(json.dumps(result, indent=2)+'\n')
    print('Feature window ends at Jan 08 12:00 UTC; first publication 12:30; correction 14:00')
    for row in result['after_backfill']:
        feature=row['feature']; value='missing' if feature is None else feature['value']
        print(f"{row['id']}: predict_at={row['predict_at']}; feature={value}")
    print('13:00 historical read unchanged after correction:', result['historical_read_unchanged'])
    print('Retained user-01 vintages:', result['retained_vintages_user_01'])


if __name__=='__main__': main()