io-pimdir 0.4.0

//! The action queue and collection generations: a producer's enqueue
//! against a real temp store, the owner's drain per action kind, parking,
//! refcount pinning of queued bodies, generation bumps, and the refusal
//! of a store stamped with an unserviced schema version.

use std::{io::Write, path::Path};

use io_pimdir::{
    PimdirBlobs, PimdirError, PimdirProducer, PimdirReader, PimdirSourceStore, PimdirStore,
    codec::PimdirAction, hash::PimdirHashAlgo, sql,
};
use io_replica::{
    change::{ReplicaDropReason, ReplicaWriteOp},
    client::ReplicaStorage,
    collection::ReplicaCollectionId,
    object::{ReplicaHash, ReplicaObject},
    placement::{
        ReplicaBase, ReplicaFlags, ReplicaHandle, ReplicaLevel, ReplicaLinkId, ReplicaMeta,
        ReplicaPlacement, ReplicaStatus,
    },
    storage::ReplicaLoadScope,
};

const NOW: &str = "2026-08-07T00:00:00Z";

fn inbox() -> ReplicaCollectionId {
    ReplicaCollectionId("INBOX".into())
}

/// A hydrated, linked placement with a matching base (so it projects clean).
fn placement(handle: &str, link: &str, hash: &str, flags: &[&str]) -> ReplicaPlacement {
    let flags = ReplicaFlags::from_iter(flags.iter().copied());
    ReplicaPlacement {
        sort_key: Default::default(),
        collection: inbox(),
        handle: ReplicaHandle(handle.into()),
        link_id: Some(ReplicaLinkId(link.into())),
        object: Some(ReplicaHash(hash.into())),
        level: ReplicaLevel::Full,
        meta: None,
        flags: flags.clone(),
        status: ReplicaStatus::Clean,
        conflict_revision: None,
        conflict_object: None,
        base: Some(ReplicaBase {
            flags,
            revision: None,
            object: Some(ReplicaHash(hash.into())),
        }),
        origin: None,
    }
}

fn store_object(hash: &str, body: &[u8]) -> ReplicaWriteOp {
    ReplicaWriteOp::StoreObject {
        object: ReplicaObject {
            hash: ReplicaHash(hash.into()),
            size: body.len(),
        },
        body: Some(body.to_vec()),
    }
}

fn blob_exists(dir: &Path, hash: &str) -> bool {
    dir.join("objects")
        .join(&hash[0..2])
        .join(&hash[2..4])
        .join(hash)
        .exists()
}

/// Seeds a one-item INBOX and returns `(store, seq)` for the item.
fn seeded(dir: &Path) -> (PimdirSourceStore, i64) {
    let mut store = PimdirStore::open(dir).unwrap().for_source("local");
    store
        .write(vec![
            store_object("cafebabe", b"abc"),
            ReplicaWriteOp::UpsertPlacement(placement("1", "mid:a", "cafebabe", &["\\Seen"])),
        ])
        .unwrap();
    let seq = store.list_items("INBOX", None, 10).unwrap()[0].seq;
    (store, seq)
}

/// Streams `body` into the blob store under `hash`, the producer's
/// blob-first step, and returns its size.
fn write_blob(dir: &Path, hash: &str, body: &[u8]) -> u64 {
    let blobs = PimdirBlobs::open(dir, PimdirHashAlgo::default());
    let mut writer = blobs.writer().unwrap();
    writer.write_all(body).unwrap();
    writer.commit(&ReplicaHash(hash.into())).unwrap()
}

#[test]
fn a_queued_add_round_trips_into_a_staged_item() {
    let dir = tempfile::tempdir().unwrap();
    let (mut store, _) = seeded(dir.path());

    // the producer writes the blob durably first, then enqueues in one
    // transaction, and the pending action shows in its own reads
    let size = write_blob(dir.path(), "beef0000", b"new body");
    let mut producer = PimdirProducer::open(dir.path(), "smtp").unwrap();
    let add = PimdirAction::Add {
        link_id: Some(ReplicaLinkId("mid:new".into())),
        flags: ReplicaFlags::from_iter(["\\Draft"]),
        object: Some(ReplicaHash("beef0000".into())),
        meta: Some(ReplicaMeta("{\"subject\":\"hi\",\"v\":1}".into())),
        handle: Some(ReplicaHandle("draft-1".into())),
    };
    producer.enqueue("INBOX", &add, Some(size), NOW).unwrap();

    let pending = producer.pending_actions("INBOX").unwrap();
    assert_eq!(pending.len(), 1);
    assert_eq!(pending[0].action, add);
    assert_eq!(pending[0].producer, "smtp");
    assert_eq!(store.queued_collections().unwrap(), ["INBOX"]);

    // the owner drains: the item lands, staged as a pending push
    let report = store.drain_collection("INBOX").unwrap();
    assert_eq!((report.applied, report.parked), (1, 0));
    assert!(store.pending_actions("INBOX").unwrap().is_empty());
    assert!(store.queued_collections().unwrap().is_empty());

    let items = store.list_items("INBOX", None, 10).unwrap();
    let item = items.iter().find(|i| i.link_id.0 == "mid:new").unwrap();
    assert_eq!(item.object, Some(ReplicaHash("beef0000".into())));
    assert_eq!(item.level, ReplicaLevel::Full);
    assert!(item.flags.contains("\\Draft"));
    assert_eq!(
        item.meta.as_ref().unwrap().0,
        "{\"subject\":\"hi\",\"v\":1}"
    );

    // projected as a base-less pending push, the shape a staged
    // in-process Add projects
    let projected = store
        .load(&inbox(), &ReplicaLoadScope::All)
        .unwrap()
        .placements;
    let created = projected
        .iter()
        .find(|p| p.link_id.as_ref().is_some_and(|l| l.0 == "mid:new"))
        .unwrap();
    assert_ne!(created.status, ReplicaStatus::Clean, "a pending push");
    assert!(created.base.is_none(), "no prior sync: an append push");
}

#[test]
fn a_duplicate_add_parks_instead_of_clobbering() {
    let dir = tempfile::tempdir().unwrap();
    let (mut store, _) = seeded(dir.path());

    let mut producer = PimdirProducer::open(dir.path(), "test").unwrap();
    let add = PimdirAction::Add {
        link_id: Some(ReplicaLinkId("mid:a".into())),
        flags: ReplicaFlags::default(),
        object: None,
        meta: None,
        handle: None,
    };
    producer.enqueue("INBOX", &add, None, NOW).unwrap();

    let report = store.drain_collection("INBOX").unwrap();
    assert_eq!((report.applied, report.parked), (0, 1));
    let parked = store.parked_actions().unwrap();
    assert_eq!(parked.len(), 1);
    assert!(parked[0].error.contains("already present"), "{parked:?}");
    assert_eq!(parked[0].attempts, 1);
    // the live item is untouched
    assert!(
        store.list_items("INBOX", None, 10).unwrap()[0]
            .flags
            .contains("\\Seen")
    );
}

#[test]
fn set_flags_is_absolute_and_reapplies_idempotently() {
    let dir = tempfile::tempdir().unwrap();
    let (mut store, seq) = seeded(dir.path());
    let mut producer = PimdirProducer::open(dir.path(), "test").unwrap();
    let set = PimdirAction::SetFlags {
        seq,
        flags: ReplicaFlags::from_iter(["\\Flagged"]),
    };

    producer.enqueue("INBOX", &set, None, NOW).unwrap();
    let report = store.drain_collection("INBOX").unwrap();
    assert_eq!((report.applied, report.parked), (1, 0));

    // absolute replacement: the old flag is gone, not merged
    let item = store.get_item("INBOX", seq).unwrap().unwrap();
    assert_eq!(item.flags, ReplicaFlags::from_iter(["\\Flagged"]));

    // reapplying the same absolute set is a no-op, never an error
    producer.enqueue("INBOX", &set, None, NOW).unwrap();
    let report = store.drain_collection("INBOX").unwrap();
    assert_eq!((report.applied, report.parked), (1, 0));
    let item = store.get_item("INBOX", seq).unwrap().unwrap();
    assert_eq!(item.flags, ReplicaFlags::from_iter(["\\Flagged"]));
}

#[test]
fn remove_hides_the_item_and_an_absent_remove_succeeds() {
    let dir = tempfile::tempdir().unwrap();
    let (mut store, seq) = seeded(dir.path());
    let mut producer = PimdirProducer::open(dir.path(), "test").unwrap();

    producer
        .enqueue("INBOX", &PimdirAction::Remove { seq }, None, NOW)
        .unwrap();
    let report = store.drain_collection("INBOX").unwrap();
    assert_eq!((report.applied, report.parked), (1, 0));

    // hidden from reads, kept as a tombstone for the sync to push
    assert!(store.get_item("INBOX", seq).unwrap().is_none());
    let projected = store
        .load(&inbox(), &ReplicaLoadScope::All)
        .unwrap()
        .placements;
    assert_eq!(projected[0].status, ReplicaStatus::Tombstone);

    // removing it again is success, not a park
    producer
        .enqueue("INBOX", &PimdirAction::Remove { seq }, None, NOW)
        .unwrap();
    let report = store.drain_collection("INBOX").unwrap();
    assert_eq!((report.applied, report.parked), (1, 0));
    assert!(store.parked_actions().unwrap().is_empty());
}

#[test]
fn copy_fills_the_target_and_move_also_empties_the_source() {
    let dir = tempfile::tempdir().unwrap();
    let mut store = PimdirStore::open(dir.path()).unwrap().for_source("local");
    store
        .write(vec![
            store_object("cafebabe", b"a"),
            store_object("cafebabf", b"b"),
            ReplicaWriteOp::UpsertPlacement(placement("1", "mid:a", "cafebabe", &[])),
            ReplicaWriteOp::UpsertPlacement(placement("2", "mid:b", "cafebabf", &[])),
        ])
        .unwrap();
    let seq_of = |store: &PimdirStore, link: &str| {
        store
            .list_items("INBOX", None, 10)
            .unwrap()
            .into_iter()
            .find(|i| i.link_id.0 == link)
            .unwrap()
            .seq
    };
    let seq_a = seq_of(&store, "mid:a");
    let seq_b = seq_of(&store, "mid:b");

    let mut producer = PimdirProducer::open(dir.path(), "test").unwrap();
    producer
        .enqueue(
            "INBOX",
            &PimdirAction::Copy {
                seq: seq_a,
                to: ReplicaCollectionId("Backup".into()),
            },
            None,
            NOW,
        )
        .unwrap();
    producer
        .enqueue(
            "INBOX",
            &PimdirAction::Move {
                seq: seq_b,
                to: ReplicaCollectionId("Archive".into()),
            },
            None,
            NOW,
        )
        .unwrap();

    let report = store.drain_collection("INBOX").unwrap();
    assert_eq!((report.applied, report.parked), (2, 0));

    // copy: the source keeps the item, the target gains it under the
    // same seq
    assert_eq!(store.count_items("INBOX").unwrap(), 1, "mid:b moved away");
    let backup = store.list_items("Backup", None, 10).unwrap();
    assert_eq!(backup.len(), 1);
    assert_eq!(backup[0].link_id.0, "mid:a");
    assert_eq!(backup[0].seq, seq_a, "a message keeps one public id");

    // move: the source empties, the target fills
    let archive = store.list_items("Archive", None, 10).unwrap();
    assert_eq!(archive.len(), 1);
    assert_eq!(archive[0].link_id.0, "mid:b");
    assert_eq!(archive[0].seq, seq_b);
    assert!(store.get_item("INBOX", seq_b).unwrap().is_none());
}

#[test]
fn update_repoints_the_body() {
    let dir = tempfile::tempdir().unwrap();
    let (mut store, seq) = seeded(dir.path());

    let size = write_blob(dir.path(), "beef0000", b"edited body");
    let mut producer = PimdirProducer::open(dir.path(), "test").unwrap();
    producer
        .enqueue(
            "INBOX",
            &PimdirAction::Update {
                seq,
                object: ReplicaHash("beef0000".into()),
                meta: None,
            },
            Some(size),
            NOW,
        )
        .unwrap();

    let report = store.drain_collection("INBOX").unwrap();
    assert_eq!((report.applied, report.parked), (1, 0));

    let item = store.get_item("INBOX", seq).unwrap().unwrap();
    assert_eq!(item.object, Some(ReplicaHash("beef0000".into())));
    // the old body survives: the sync base still references it
    assert!(blob_exists(dir.path(), "cafebabe"));
    assert!(blob_exists(dir.path(), "beef0000"));
    // projected dirty, so the next sync pushes the edit
    let projected = store
        .load(&inbox(), &ReplicaLoadScope::All)
        .unwrap()
        .placements;
    assert_eq!(projected[0].status, ReplicaStatus::Dirty);
}

#[test]
fn a_parked_action_does_not_block_later_actions() {
    let dir = tempfile::tempdir().unwrap();
    let (mut store, seq) = seeded(dir.path());
    let mut producer = PimdirProducer::open(dir.path(), "test").unwrap();

    // an unappliable action, then a valid one behind it
    producer
        .enqueue("INBOX", &PimdirAction::Remove { seq: 9999 }, None, NOW)
        .unwrap();
    // NOTE: remove-of-absent succeeds, so the parking case is set-flags.
    producer
        .enqueue(
            "INBOX",
            &PimdirAction::SetFlags {
                seq: 9999,
                flags: ReplicaFlags::default(),
            },
            None,
            NOW,
        )
        .unwrap();
    producer
        .enqueue(
            "INBOX",
            &PimdirAction::SetFlags {
                seq,
                flags: ReplicaFlags::from_iter(["\\Answered"]),
            },
            None,
            NOW,
        )
        .unwrap();

    let report = store.drain_collection("INBOX").unwrap();
    assert_eq!((report.applied, report.parked), (2, 1));

    // the later action applied despite the parked one before it
    let item = store.get_item("INBOX", seq).unwrap().unwrap();
    assert!(item.flags.contains("\\Answered"));

    // the parked row is recorded, queryable and skipped by the next pass
    let parked = store.parked_actions().unwrap();
    assert_eq!(parked.len(), 1);
    assert_eq!(parked[0].collection, "INBOX");
    assert_eq!(parked[0].action, "set-flags");
    assert!(parked[0].error.contains("unknown seq"), "{parked:?}");
    let report = store.drain_collection("INBOX").unwrap();
    assert_eq!((report.applied, report.parked), (0, 0));
    assert_eq!(store.parked_actions().unwrap().len(), 1, "never deleted");
}

#[test]
fn gc_never_sweeps_a_queued_body() {
    let dir = tempfile::tempdir().unwrap();
    let (mut store, seeded_seq) = seeded(dir.path());

    // a producer stages a body and enqueues the add referencing it
    let size = write_blob(dir.path(), "beef0000", b"queued body");
    let mut producer = PimdirProducer::open(dir.path(), "test").unwrap();
    producer
        .enqueue(
            "INBOX",
            &PimdirAction::Add {
                link_id: Some(ReplicaLinkId("mid:new".into())),
                flags: ReplicaFlags::default(),
                object: Some(ReplicaHash("beef0000".into())),
                meta: None,
                handle: Some(ReplicaHandle("draft-1".into())),
            },
            Some(size),
            NOW,
        )
        .unwrap();

    // a collector cannot run while the producer holds its staging lock:
    // the body it just wrote is pinned by nothing until the enqueue
    // commits, and that window is why the lock exists
    assert!(matches!(
        store.collect_garbage(),
        Err(PimdirError::Staging(_))
    ));
    drop(producer);

    // the owner retires and purges the seeded item, orphaning its body,
    // and collects: the queued body survives the sweep, pinned by its row
    store
        .write(vec![ReplicaWriteOp::DropPlacement {
            collection: inbox(),
            handle: ReplicaHandle("1".into()),
            reason: ReplicaDropReason::Deleted,
        }])
        .unwrap();
    assert!(store.purge(&inbox(), seeded_seq).unwrap());

    let collected = store.collect_garbage().unwrap();
    assert_eq!((collected.objects, collected.blobs), (1, 1));
    assert!(!blob_exists(dir.path(), "cafebabe"), "the orphan is taken");
    assert!(
        blob_exists(dir.path(), "beef0000"),
        "the queued body is not"
    );

    // draining hands the pin over to the applied item
    let report = store.drain_collection("INBOX").unwrap();
    assert_eq!((report.applied, report.parked), (1, 0));
    assert!(blob_exists(dir.path(), "beef0000"), "now the item pins it");

    // the hand-over was exact, so retiring and purging the item's only
    // placement orphans the body and the next collection takes it, with
    // no leaked queue pin
    let seq = store.seq_for_link("INBOX", "mid:new").unwrap().unwrap();
    store
        .write(vec![ReplicaWriteOp::DropPlacement {
            collection: inbox(),
            handle: ReplicaHandle("draft-1".into()),
            reason: ReplicaDropReason::Deleted,
        }])
        .unwrap();
    assert!(store.purge(&inbox(), seq).unwrap());
    assert_eq!(store.collect_garbage().unwrap().blobs, 1);
    assert!(!blob_exists(dir.path(), "beef0000"), "no refcount leak");
}

#[test]
fn an_unknown_kind_is_skipped_and_never_blocks_the_queue() {
    // the store defines no semantics for a `submit` intent but another
    // owner does, so skipping leaves it pending for that owner, where
    // parking would claim nobody can ever apply it
    let dir = tempfile::tempdir().unwrap();
    let (mut store, seq) = seeded(dir.path());
    let mut producer = PimdirProducer::open(dir.path(), "himalaya").unwrap();

    let size = write_blob(dir.path(), "beef0000", b"a message to send");
    let submit = PimdirAction::Unknown {
        kind: "submit".into(),
        payload: "{\"v\":1,\"object\":\"beef0000\",\"to\":[\"a@b.c\"]}".into(),
        object_hash: Some(ReplicaHash("beef0000".into())),
    };
    let id = producer.enqueue("INBOX", &submit, Some(size), NOW).unwrap();
    producer
        .enqueue(
            "INBOX",
            &PimdirAction::SetFlags {
                seq,
                flags: ReplicaFlags::from_iter(["\\Answered"]),
            },
            None,
            NOW,
        )
        .unwrap();

    let report = store.drain_collection("INBOX").unwrap();
    assert_eq!((report.applied, report.parked, report.skipped), (1, 0, 1));

    // the action behind it applied all the same
    assert!(
        store
            .get_item("INBOX", seq)
            .unwrap()
            .unwrap()
            .flags
            .contains("\\Answered")
    );

    // the intent is still pending, never parked, readable whole by the
    // owner that can perform it, and its body still pinned
    assert!(store.parked_actions().unwrap().is_empty());
    let pending = store.pending_actions("INBOX").unwrap();
    assert_eq!(pending.len(), 1);
    assert_eq!(pending[0].id, id);
    assert_eq!(pending[0].action, submit);
    assert!(blob_exists(dir.path(), "beef0000"));

    // a second pass skips it again rather than accumulating attempts
    let report = store.drain_collection("INBOX").unwrap();
    assert_eq!((report.applied, report.parked, report.skipped), (0, 0, 1));
    assert_eq!(store.pending_actions("INBOX").unwrap()[0].attempts, 0);
}

#[test]
fn an_acknowledged_action_releases_its_queued_body() {
    // the other half of skip-not-park: the owner that performed the
    // intent out of band takes the row away, and its body's pin with it
    let dir = tempfile::tempdir().unwrap();
    let (mut store, _) = seeded(dir.path());
    let mut producer = PimdirProducer::open(dir.path(), "himalaya").unwrap();

    let size = write_blob(dir.path(), "beef0000", b"sent already");
    let id = producer
        .enqueue(
            "INBOX",
            &PimdirAction::Unknown {
                kind: "submit".into(),
                payload: "{\"v\":1,\"object\":\"beef0000\"}".into(),
                object_hash: Some(ReplicaHash("beef0000".into())),
            },
            Some(size),
            NOW,
        )
        .unwrap();
    assert!(blob_exists(dir.path(), "beef0000"));

    assert!(store.drop_action(id).unwrap());
    assert!(store.pending_actions("INBOX").unwrap().is_empty());
    assert!(blob_exists(dir.path(), "beef0000"), "no write collects");

    drop(producer);
    assert_eq!(store.collect_garbage().unwrap().blobs, 1);
    assert!(
        !blob_exists(dir.path(), "beef0000"),
        "the pin went with the row"
    );

    // acknowledging a row that is already gone reports nothing to drop
    assert!(!store.drop_action(id).unwrap());
}

#[test]
fn a_failed_action_retries_until_it_is_parked() {
    // the two failure shapes an owner reports itself: transient, still
    // pending with attempts advancing, and permanent, parked with the
    // reason
    let dir = tempfile::tempdir().unwrap();
    let (mut store, seq) = seeded(dir.path());
    let mut producer = PimdirProducer::open(dir.path(), "test").unwrap();
    let id = producer
        .enqueue("INBOX", &PimdirAction::Remove { seq }, None, NOW)
        .unwrap();

    store.fail_action(id, None).unwrap();
    store.fail_action(id, None).unwrap();
    let pending = store.pending_actions("INBOX").unwrap();
    assert_eq!(pending.len(), 1, "a transient failure stays pending");
    assert_eq!(pending[0].attempts, 2);

    store.fail_action(id, Some("the channel is gone")).unwrap();
    assert!(store.pending_actions("INBOX").unwrap().is_empty());
    let parked = store.parked_actions().unwrap();
    assert_eq!(parked.len(), 1);
    assert_eq!(parked[0].id, id);
    assert_eq!(parked[0].attempts, 3, "the parking attempt counts too");
    assert_eq!(parked[0].error, "the channel is gone");

    // a parked row is still cancellable, and an unknown id is a no-op
    assert!(store.drop_action(id).unwrap());
    assert!(store.parked_actions().unwrap().is_empty());
    store.fail_action(id, Some("gone")).unwrap();
}

#[test]
fn a_generation_bump_is_visible_to_a_reader() {
    let dir = tempfile::tempdir().unwrap();
    let mut store = PimdirStore::open(dir.path()).unwrap().for_source("local");
    store.ensure_collection("INBOX", "message/rfc822").unwrap();

    // the epoch starts at 1 and ordinary writes never bump it
    assert_eq!(store.generation("INBOX").unwrap(), Some(1));
    store
        .write(vec![
            store_object("cafebabe", b"abc"),
            ReplicaWriteOp::UpsertPlacement(placement("1", "mid:a", "cafebabe", &[])),
        ])
        .unwrap();
    assert_eq!(store.generation("INBOX").unwrap(), Some(1));

    // a rekey bumps it in the same transaction: the item is re-bound
    // under its new handle and the epoch advances together
    let mut rekeyed = placement("101", "mid:a", "cafebabe", &[]);
    rekeyed.status = ReplicaStatus::Clean;
    let generation = store
        .write_rekeyed(
            "INBOX",
            vec![
                ReplicaWriteOp::DropPlacement {
                    collection: inbox(),
                    handle: ReplicaHandle("1".into()),
                    // NOTE: a rekey replaces the row rather than deleting
                    // the item; the same batch upserts it under its new
                    // handle.
                    reason: ReplicaDropReason::Superseded,
                },
                ReplicaWriteOp::UpsertPlacement(rekeyed),
            ],
        )
        .unwrap();
    assert_eq!(generation, 2);

    // and the item came with it: without this, the drop and the upsert
    // read as one source reporting an identity under a second handle, so
    // the binding would keep the handle the server had just voided
    let carried = store.load(&inbox(), &ReplicaLoadScope::All).unwrap();
    assert_eq!(carried.placements.len(), 1);
    assert_eq!(carried.placements[0].handle, ReplicaHandle("101".into()));

    // a second reader handle over the same files observes the new epoch
    let reader = PimdirStore::open(dir.path()).unwrap();
    assert_eq!(reader.generation("INBOX").unwrap(), Some(2));
    let collections = reader.list_collections().unwrap();
    assert_eq!(collections[0].generation, 2);
    // an unknown collection has no epoch
    assert_eq!(reader.generation("nope").unwrap(), None);
}

#[test]
fn a_store_stamped_with_a_higher_version_is_refused() {
    let dir = tempfile::tempdir().unwrap();

    // a fresh store lands at the current draft schema version, with
    // `user_version` and `store_meta.version` in agreement
    {
        let _store = PimdirStore::open(dir.path()).unwrap();
        let conn = rusqlite::Connection::open(dir.path().join("pimdir.db")).unwrap();
        let user_version: i64 = conn
            .pragma_query_value(None, "user_version", |r| r.get(0))
            .unwrap();
        let meta_version: i64 = conn
            .query_row("SELECT version FROM store_meta WHERE id = 1", [], |r| {
                r.get(0)
            })
            .unwrap();
        assert_eq!((user_version, meta_version), (sql::VERSION, sql::VERSION));
    }

    // stamp the store with a higher schema version, as a newer draft
    // would
    {
        let conn = rusqlite::Connection::open(dir.path().join("pimdir.db")).unwrap();
        conn.pragma_update(None, "user_version", sql::VERSION + 1)
            .unwrap();
    }

    // every opener refuses it rather than half-reading: the owner, which
    // never migrates, the producer, and the read-only reader, which
    // refuses any version but the current one
    for result in [
        PimdirStore::open(dir.path()).map(drop),
        PimdirProducer::open(dir.path(), "test").map(drop),
        PimdirReader::open(dir.path()).map(drop),
    ] {
        match result {
            Err(PimdirError::Version { found }) => assert_eq!(found, sql::VERSION + 1),
            Err(other) => panic!("expected a version refusal, got {other:?}"),
            Ok(()) => panic!("expected a version refusal, got an open store"),
        }
    }

    // a producer also refuses a store the owner has not created yet: it
    // never creates the schema
    let fresh = tempfile::tempdir().unwrap();
    rusqlite::Connection::open(fresh.path().join("pimdir.db")).unwrap();
    match PimdirProducer::open(fresh.path(), "test") {
        Err(PimdirError::Uncreated) => {}
        Err(other) => panic!("expected an uncreated-store refusal, got {other:?}"),
        Ok(_) => panic!("expected an uncreated-store refusal, got a producer"),
    }
}

/// The spec's "cannot apply *here*" case, which is what neverest hit on a
/// mail account also syncing contacts and calendar: its sources take turns
/// draining, and the first one to reach a mail action holds no binding for
/// the item it names. Parking there loses the action for the source that
/// could have applied it, since a parked row is skipped by every later
/// drain, so the refusal has to leave the row alone instead.
#[test]
fn an_action_the_draining_source_cannot_place_is_skipped_not_parked() {
    let dir = tempfile::tempdir().unwrap();
    let (mut owner, seq) = seeded(dir.path());

    let mut producer = PimdirProducer::open(dir.path(), "frontend").unwrap();
    producer
        .enqueue(
            "INBOX",
            &PimdirAction::SetFlags {
                seq,
                flags: ReplicaFlags::default(),
            },
            None,
            NOW,
        )
        .unwrap();

    // A source with no binding in this collection drains first.
    let mut stranger = PimdirStore::open(dir.path()).unwrap().for_source("other");
    let report = stranger.drain_collection("INBOX").unwrap();
    assert_eq!((report.applied, report.skipped, report.parked), (0, 1, 0));
    assert!(stranger.parked_actions().unwrap().is_empty());

    // The row was left exactly as it was found, so the source that holds
    // the item still applies it.
    let report = owner.drain_collection("INBOX").unwrap();
    assert_eq!((report.applied, report.skipped, report.parked), (1, 0, 0));

    let item = owner.get_item("INBOX", seq).unwrap().unwrap();
    assert_eq!(item.flags, ReplicaFlags::default());
}