Record placement decisions for replicas and provider transitions (#18)

ddiff --git a/server/events.js b/server/events.js
iindex c45c9c9..8a4c973 100644
---- a/server/events.js
++++ b/server/events.js
@@@ -42,6 +42,13 @@ const TYPES = {
      'storage.recovered': ['media.storage.recovered', 'storage'],
  };
  
++// Placement decisions have no webhook twin; callers record them in the outbox transaction.
++const PLACEMENT_TYPES = new Set([
++    'media.replica.requested', 'media.replica.ready', 'media.replica.draining', 'media.replica.evicted',
++    'media.object.hot', 'media.region.hot', 'media.delivery.surge', 'media.delivery.degraded',
++    'provider.health.degraded', 'provider.capacity.warning', 'provider.cost.threshold',
++]);
++
  // media.job.<transition>: progress is low priority, the outcome (and a proposal waiting for its owner) important.
  const JOB_TRANSITIONS = {
      proposed: 'important', queued: 'low', started: 'low', retrying: 'low',
@@@ -117,6 +124,12 @@ async function record(webhookEvent, appId, data) {
      return await enqueue(eventType, appId, subject, data, eventType === 'media.storage.recovered' ? 'low' : 'important');
  }
  
++/** Queue one placement decision inside its transaction; disabled and sandbox outboxes return null. */
++async function recordPlacement(eventType, appId, subject, payload, { priority = 'low' } = {}) {
++    if (!PLACEMENT_TYPES.has(eventType)) throw new Error(`unknown placement event ${eventType}`);
++    return await enqueue(eventType, appId, subject, payload, priority);
++}
++
  /**
   * Queue media.job.<transition> for a job (queue.jobEvent: ids, state, counters, ISO times and
   * has_result; never params, result, error text, idempotency key or who created/decided it; the
@@@ -274,4 +287,4 @@ function _reset() {
      stats.queued = 0; stats.lastError = null;
  }
  
--module.exports = { init, initWriter, record, recordIndexDocument, recordModeration, recordJob, recordObjectChanges, drainObjectChanges, discardObjectChanges, objectChangeEvent, kick, status, TYPES, JOB_TRANSITIONS, _reset };
++module.exports = { init, initWriter, record, recordPlacement, recordIndexDocument, recordModeration, recordJob, recordObjectChanges, drainObjectChanges, discardObjectChanges, objectChangeEvent, kick, status, TYPES, PLACEMENT_TYPES, JOB_TRANSITIONS, _reset };
ddiff --git a/server/objects/tiering.js b/server/objects/tiering.js
iindex 281ba77..f9acf41 100644
---- a/server/objects/tiering.js
++++ b/server/objects/tiering.js
@@@ -52,6 +52,8 @@ const LATENCY_VALUE = 1;             // $ a request served from R2 is worth: no
  
  const vodStorage = () => require('../vod/vod-storage');
  const model = () => require('./model');
++const events = () => require('../events');
++const subjectOf = (obj) => ({ type: 'object', id: obj.id });
  
  const sweepState = { running: false, lastRunAt: null, lastResult: null };
  
@@@ -227,6 +229,13 @@ async function promote(objectId, ctx = {}) {
      if (!settings.active) return await log('dry_run', `gate off (active = false): would promote: ${why}${vs.providerAvailable('r2') ? '' : ' (R2 is not available now)'}`);
      if (!vs.providerAvailable('r2')) return await log('refused', `${why}; refused: R2 is not available`, { error: 'R2 not available' });
  
++    await db.getDb().tx(async () => {
++        await events().recordPlacement('media.replica.requested', obj.app_id, subjectOf(obj), {
++            object_id: obj.id, app_id: obj.app_id || null, class: storagePolicy.classOf(obj), action: 'promote',
++            from: canon ? canon.provider : null, to: 'r2', provider: 'r2',
++        });
++    });
++    events().kick();
      const check = await confirmCanonical(obj, canon);
      if (!check.ok) return await log('refused', `${why}; refused: ${check.why}`, { error: check.why });
      const key = r2KeyFor(obj, canon);
@@@ -257,7 +266,12 @@ async function promote(objectId, ctx = {}) {
          await model().upsertLocation(obj.id, { provider: 'r2', bucket: vs.bucketFor('r2'), key, storage_class: 'cache', state: 'present',
              checksum: obj.content_hash, size_bytes: size, verified: true });
          await model().setLocationState(canon.id, { state: 'present', size_bytes: size });
++        await events().recordPlacement('media.replica.ready', obj.app_id, subjectOf(obj), {
++            object_id: obj.id, app_id: obj.app_id || null, class: storagePolicy.classOf(obj), action: 'promote',
++            provider: 'r2', key, bytes: size, replica_since: new Date().toISOString(),
++        });
      });
++    events().kick();
      console.log(`[ObjectTiers] ${obj.id} (${obj.app_id}) promoted to R2: ${key} (${(size / MB).toFixed(1)} MB)`);
      return await log('done', why, { bytes: size });
  }
@@@ -282,6 +296,13 @@ async function demote(objectId, ctx = {}) {
      // The last good copy is never the one removed: the canonical copy must check out now.
      const check = await confirmCanonical(obj, canon);
      if (!check.ok) return await log('refused', `${why}; refused: ${check.why}; the R2 copy is kept (it may be the last good copy)`, { error: check.why });
++    await db.getDb().tx(async () => {
++        await events().recordPlacement('media.replica.draining', obj.app_id, subjectOf(obj), {
++            object_id: obj.id, app_id: obj.app_id || null, class: storagePolicy.classOf(obj), action: 'demote',
++            provider: 'r2', key: r2.key,
++        });
++    });
++    events().kick();
      try {
          await vs.deleteObject('r2', r2.key);
          const still = await vs.headObject('r2', r2.key);
@@@ -292,7 +313,12 @@ async function demote(objectId, ctx = {}) {
      await db.getDb().tx(async () => {
          await db.run('DELETE FROM media_locations WHERE id = ?', [r2.id]);
          await model().setLocationState(canon.id, { state: 'present', size_bytes: Number(obj.size_bytes) });
++        await events().recordPlacement('media.replica.evicted', obj.app_id, subjectOf(obj), {
++            object_id: obj.id, app_id: obj.app_id || null, class: storagePolicy.classOf(obj), action: 'demote',
++            provider: 'r2', key: r2.key,
++        });
      });
++    events().kick();
      console.log(`[ObjectTiers] ${obj.id} (${obj.app_id}) demoted from R2 (${why})`);
      return await log('done', why);
  }
ddiff --git a/server/placement/providers.js b/server/placement/providers.js
iindex 8f233db..9491b09 100644
---- a/server/placement/providers.js
++++ b/server/placement/providers.js
@@@ -57,6 +57,14 @@ const state = {
      lastHealthError: {},   // provider -> string
  };
  
++async function announceProviderEvent(eventType, provider, payload) {
++    try {
++        await require('../db/database').getDb().tx(async () =>
++            await require('../events').recordPlacement(eventType, null, { type: 'provider', id: provider }, { provider, ...payload }));
++        require('../events').kick();
++    } catch (err) { console.warn(`[Placement] ${eventType} not announced: ${err.message}`); }
++}
++
  /**
   * Capability gating. 'report' (the default) records what each provider can do and exposes it (health, the
   * admin placement view) without taking anything out of routing; 'enforce' (MEDIA_CAPABILITY_GATE=enforce) removes
@@@ -105,6 +113,16 @@ function capabilities(name) {
  /** Record the latest probe result in memory. */
  function setCapabilities(name, caps) {
      if (!state.capabilities) state.capabilities = {};
++    const old = state.capabilities[name];
++    const classesOf = (snapshot) => snapshot.classes || providerClasses(name).filter((cls) =>
++        CLASS_REQUIREMENTS[cls].every((cap) => snapshot.capabilities && snapshot.capabilities[cap]));
++    if (old && old.passed && caps.passed) {
++        const prev = classesOf(old);
++        const next = classesOf(caps);
++        if (next.length < prev.length) {
++            void announceProviderEvent('provider.capacity.warning', name, { classes: next, previous_classes: prev });
++        }
++    }
      state.capabilities[name] = caps;
  }
  
@@@ -228,8 +246,10 @@ async function liveHealth(name) {
          observe(name, 'head', true, ms);
          return { configured: true, healthy: true, lastCheck: state.lastHealthCheck[name], latencyMs: ms };
      } catch (err) {
++        const wasHealthy = state.lastHealthCheck[name] != null && state.lastHealthError[name] == null;
          state.lastHealthCheck[name] = Date.now();
          state.lastHealthError[name] = err.message || String(err);
++        if (wasHealthy) void announceProviderEvent('provider.health.degraded', name, { healthy: false, last_error: state.lastHealthError[name] });
          const ms = state.lastHealthCheck[name] - started;
          observe(name, 'head', false, ms);
          return { configured: true, healthy: false, lastCheck: state.lastHealthCheck[name], error: state.lastHealthError[name], latencyMs: ms };
@@@ -300,4 +320,4 @@ module.exports = {
      startHealthLoop, stop, reset,
      // Test-only setter (used by tests to inject a known capability snapshot without touching the SDK).
      _setCapabilities: setCapabilities,
--};
\\ No newline at end of file
++};
ddiff --git a/test/placement-events.test.js b/test/placement-events.test.js
nnew file mode 100644
iindex 0000000..3813e5f
---- /dev/null
++++ b/test/placement-events.test.js
@@@ -0,0 +1,169 @@
++'use strict';
++// Replica decisions and provider transitions use the durable placement outbox.
++const assert = require('assert');
++const crypto = require('crypto');
++const fs = require('fs');
++const http = require('http');
++const os = require('os');
++const path = require('path');
++
++(async () => {
++    const tmp = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-media-placement-events-'));
++    const dir = (name) => { const p = path.join(tmp, name); fs.mkdirSync(p, { recursive: true }); return p; };
++    Object.assign(process.env, {
++        VOD_PATH: dir('vods'), CLIPS_PATH: dir('clips'), FILES_PATH: dir('files'),
++        THUMBNAILS_PATH: dir('thumbnails'), PASTES_PATH: dir('pastes'), OBJECTS_PATH: dir('objects'),
++    });
++    const stub = http.createServer((req, res) => {
++        req.resume();
++        req.on('end', () => {
++            res.setHeader('Content-Type', 'application/json');
++            if (req.url === '/oauth/token') return res.end(JSON.stringify({ access_token: 'tok', token_type: 'Bearer', expires_in: 300 }));
++            res.statusCode = 503; res.end('{}'); // Keep every envelope in the outbox for inspection.
++        });
++    });
++    await new Promise((resolve) => stub.listen(0, '127.0.0.1', resolve));
++
++    const db = require('../server/db/database');
++    const model = require('../server/objects/model');
++    const storage = require('../server/vod/vod-storage');
++    const policy = require('../server/objects/tier-policy');
++    const tiering = require('../server/objects/tiering');
++    const providers = require('../server/placement/providers');
++    const events = require('../server/events');
++    const base = `http://127.0.0.1:${stub.address().port}`;
++    const copies = new Map();
++    let corrupt = false;
++    storage.providerAvailable = () => true;
++    storage.providerConfigured = () => true;
++    storage.bucketFor = () => 'r2-bucket';
++    storage.uploadFile = async (_provider, key, file) => {
++        const bytes = fs.readFileSync(file);
++        if (corrupt) bytes[0] ^= 0xff;
++        copies.set(key, bytes);
++    };
++    storage.headObject = async (_provider, key) => copies.has(key) ? { size: copies.get(key).length } : null;
++    storage.sha256Object = async (_provider, key) => copies.has(key)
++        ? { sha256: crypto.createHash('sha256').update(copies.get(key)).digest('hex') } : null;
++    storage.deleteObject = async (_provider, key) => { copies.delete(key); };
++
++    await policy.init();
++    await db.upsertApp({ app_id: 'live', api_key: 'live-key-placement-events' });
++    const sandbox = (await db.ensureProjectTenant('prj_01TESTPLACEVENTS000000001', 'sandbox', 1024)).app_id;
++    const enabled = { ...policy.settings(), active: true };
++    const disabled = { ...enabled, active: false };
++    const raw = db.getDb();
++    const placementRows = async () => (await raw.prepare('SELECT envelope FROM event_outbox ORDER BY id').all())
++        .map((row) => typeof row.envelope === 'string' ? JSON.parse(row.envelope) : row.envelope)
++        .filter((event) => events.PLACEMENT_TYPES.has(event.event_type));
++    const rowsFor = async (id) => (await placementRows()).filter((event) => event.subject.id === id);
++    const waitFor = async (test) => {
++        for (let i = 0; i < 100; i++) {
++            if (await test()) return;
++            await new Promise((resolve) => setTimeout(resolve, 10));
++        }
++        throw new Error('timed out waiting for a placement event');
++    };
++    const addObject = async (app = 'live') => {
++        const bytes = Buffer.from('a verified video object for placement events');
++        const checksum = crypto.createHash('sha256').update(bytes).digest('hex');
++        const id = await model.createObject({ app_id: app, kind: 'file', visibility: 'public', lifecycle_status: 'ready',
++            mime_type: 'video/mp4', size_bytes: bytes.length, content_hash: checksum, canonical_provider: 'local' });
++        const file = path.join(process.env.OBJECTS_PATH, app, id);
++        fs.mkdirSync(path.dirname(file), { recursive: true });
++        fs.writeFileSync(file, bytes);
++        await model.updateObject(id, { canonical_key: file });
++        await model.upsertLocation(id, { provider: 'local', key: file, state: 'present', size_bytes: bytes.length, checksum, verified: true });
++        return { id, bytes };
++    };
++
++    assert.ok(events.init({ eventsUrl: base, clientSecret: 's', networkUrl: base, intervalMs: 3600000 }));
++    assert.strictEqual(events.PLACEMENT_TYPES.size, 11);
++    await assert.rejects(events.recordPlacement('not.a.placement', null, { type: 'provider', id: 'r2' }, {}), /unknown placement event/);
++
++    // 1. Promotion records the intent before copying, then the verified replica with its exact payload.
++    const promoted = await addObject();
++    assert.strictEqual((await tiering.promote(promoted.id, { settings: enabled })).outcome, 'done');
++    let rows = await rowsFor(promoted.id);
++    assert.deepStrictEqual(rows.map((e) => e.event_type), ['media.replica.requested', 'media.replica.ready']);
++    for (const e of rows) {
++        assert.deepStrictEqual(e.subject, { type: 'object', id: promoted.id });
++        assert.deepStrictEqual([e.priority, e.visibility, e.payload.class], ['low', 'internal', 'video']);
++    }
++    assert.deepStrictEqual(Object.keys(rows[0].payload).sort(), ['action', 'app_id', 'class', 'from', 'object_id', 'provider', 'to']);
++    assert.deepStrictEqual(Object.keys(rows[1].payload).sort(), ['action', 'app_id', 'bytes', 'class', 'key', 'object_id', 'provider', 'replica_since']);
++    assert.deepStrictEqual([rows[0].payload.from, rows[0].payload.to, rows[1].payload.bytes], ['local', 'r2', promoted.bytes.length]);
++    console.log('✅ promote: requested then ready, with object subject and verified video payload');
++
++    // 2. Gate off is a dry run and writes no placement event.
++    const dry = await addObject();
++    assert.strictEqual((await tiering.promote(dry.id, { settings: disabled })).outcome, 'dry_run');
++    assert.deepStrictEqual(await rowsFor(dry.id), []);
++    console.log('✅ gate off: dry_run without placement events');
++
++    // 3. A bad copied checksum leaves only the request.
++    const failed = await addObject();
++    corrupt = true;
++    assert.strictEqual((await tiering.promote(failed.id, { settings: enabled })).outcome, 'failed');
++    corrupt = false;
++    assert.deepStrictEqual((await rowsFor(failed.id)).map((e) => e.event_type), ['media.replica.requested']);
++    console.log('✅ failed verification: requested without ready');
++
++    // 4. Demotion announces draining before deletion and eviction after the location is removed.
++    assert.strictEqual((await tiering.demote(promoted.id, { settings: enabled })).outcome, 'done');
++    rows = await rowsFor(promoted.id);
++    assert.deepStrictEqual(rows.map((e) => e.event_type),
++        ['media.replica.requested', 'media.replica.ready', 'media.replica.draining', 'media.replica.evicted']);
++    for (const e of rows.slice(2)) {
++        assert.deepStrictEqual(e.subject, { type: 'object', id: promoted.id });
++        assert.deepStrictEqual(Object.keys(e.payload).sort(), ['action', 'app_id', 'class', 'key', 'object_id', 'provider']);
++        assert.deepStrictEqual([e.payload.action, e.payload.class, e.payload.provider], ['demote', 'video', 'r2']);
++    }
++    console.log('✅ demote: draining then evicted');
++
++    // 5. Sandbox placement is refused and has no outbox row.
++    const sandboxed = await addObject(sandbox);
++    assert.strictEqual((await tiering.promote(sandboxed.id, { settings: enabled })).outcome, 'refused');
++    assert.strictEqual(await raw.tx(async () => await events.recordPlacement('media.replica.requested', sandbox,
++        { type: 'object', id: sandboxed.id }, { object_id: sandboxed.id })), null);
++    assert.deepStrictEqual(await rowsFor(sandboxed.id), []);
++    console.log('✅ sandbox tenant: zero placement rows');
++
++    // 6. With the outbox disabled, a normal move still succeeds and writes no event.
++    events._reset();
++    assert.strictEqual(events.init({ eventsUrl: '', clientSecret: '' }), null);
++    const off = await addObject();
++    assert.strictEqual((await tiering.promote(off.id, { settings: enabled })).outcome, 'done');
++    assert.deepStrictEqual(await rowsFor(off.id), []);
++    console.log('✅ outbox off: successful move and zero placement rows');
++
++    // 7. A healthy-to-unhealthy flip and a capability loss each emit once, not once per tick.
++    events._reset();
++    events.init({ eventsUrl: base, clientSecret: 's', networkUrl: base, intervalMs: 3600000 });
++    providers.reset();
++    let healthy = true;
++    storage.probeProvider = async () => { if (!healthy) throw new Error('test outage'); };
++    assert.strictEqual((await providers.liveHealth('r2')).healthy, true);
++    healthy = false;
++    assert.strictEqual((await providers.liveHealth('r2')).healthy, false);
++    await waitFor(async () => (await rowsFor('r2')).some((e) => e.event_type === 'provider.health.degraded'));
++    await providers.liveHealth('r2');
++    assert.strictEqual((await rowsFor('r2')).filter((e) => e.event_type === 'provider.health.degraded').length, 1);
++    providers._setCapabilities('r2', { passed: true, capabilities: { range: true, range_206: true } });
++    providers._setCapabilities('r2', { passed: true, capabilities: { range: true, range_206: false } });
++    await waitFor(async () => (await rowsFor('r2')).some((e) => e.event_type === 'provider.capacity.warning'));
++    providers._setCapabilities('r2', { passed: true, capabilities: { range: true, range_206: false } });
++    rows = await rowsFor('r2');
++    assert.deepStrictEqual(rows.map((e) => e.event_type), ['provider.health.degraded', 'provider.capacity.warning']);
++    assert.deepStrictEqual(rows.map((e) => e.subject), [{ type: 'provider', id: 'r2' }, { type: 'provider', id: 'r2' }]);
++    assert.deepStrictEqual(rows[1].payload, { app_id: null, provider: 'r2', classes: [], previous_classes: ['online-hot'] });
++    console.log('✅ provider transitions: one health degradation and one class-loss warning');
++
++    providers.reset();
++    events._reset();
++    stub.close();
++    await db.close();
++    fs.rmSync(tmp, { recursive: true, force: true });
++    console.log('placement-events: all seven cases passed');
++    process.exit(0);
++})().catch((err) => { console.error(err); process.exit(1); });