later 0.0.44

//! Replayable records stored in Later's existing topic partitions.
//!
//! Enable the `retained-log` feature together with `sqlite` or `postgres`.
//! A log topic uses the same registered [`crate::topic::TopicConfig`], stable
//! partition-key hash, and fixed partition count as sequential jobs. Log
//! offsets are separate from job sequences, so completing a job never removes
//! a retained record.
//!
//! # Normal topics vs. compacted topics
//!
//! Both kinds of topic share the exact same API — `register_topic`,
//! `append`, and `read` on `SqliteLog`/`PostgresLog`. The difference is
//! entirely in how you pick keys and which cleanup method you call
//! afterwards:
//!
//! - **Normal (append-only) topic**: give every record its own distinct
//!   [`crate::topic::PartitionKey`] (or don't care about collisions). Every appended
//!   record survives until you explicitly trim it with `delete_before`,
//!   which unconditionally deletes everything below a given offset once
//!   every registered consumer group has committed past it. Use this for an
//!   event log, an audit trail, or anything where every event matters, not
//!   just the latest one for a given key.
//!
//! - **Compacted topic**: reuse the *same* key for every record that is an
//!   update to one logical entity (for example, the key is an order ID and
//!   each record is that order's latest state). Periodically call
//!   `compact_partition` once per partition — it deletes an older record
//!   only when a strictly newer record with the *same key* exists **and**
//!   every registered consumer group has already committed past it, the
//!   same rule Kafka compacted topics use. A key written only once is never
//!   touched by compaction (only `delete_before` removes it); records under
//!   different keys never affect each other's compaction.
//!
//! Compaction is never automatic — nothing calls `compact_partition` for
//! you, so run it yourself (for example from a periodic maintenance task).
//! Both cleanup methods derive their safe watermark from the *minimum*
//! offset committed across every registered consumer group on a partition,
//! so a group that stops committing blocks cleanup forever; remove
//! abandoned members with `leave_group`.

use crate::topic::{partition_for_key, PartitionId, PartitionKey, TopicConfig};

/// One immutable record in a topic partition.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct LogRecord {
    /// The topic that contains this record.
    pub topic: String,
    /// The partition selected from [`Self::key`].
    pub partition: PartitionId,
    /// The monotonic offset within `topic` and `partition`.
    pub offset: i64,
    /// The routing key used to choose this partition.
    pub key: PartitionKey,
    /// Application-defined record bytes.
    pub payload: Vec<u8>,
    /// Time when the producer committed this record.
    pub created_at: crate::UtcDateTime,
}

/// How far behind one consumer group is on one topic partition.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct ConsumerLag {
    /// The partition this reading is for.
    pub partition: PartitionId,
    /// The consumer group this reading is for.
    pub group: String,
    /// The offset the next `append` on this partition would receive - one
    /// past the newest written record.
    pub end_offset: i64,
    /// This group's own [`SqliteLog::committed_offset`]/
    /// [`PostgresLog::committed_offset`] - one past the newest record it has
    /// committed.
    pub committed_offset: i64,
    /// `end_offset - committed_offset`: records appended but not yet
    /// committed by this group. Both offsets share the same "next offset"
    /// convention, so this is never negative.
    pub lag: i64,
}

/// A read-only view onto [`SqliteLog::consumer_lag`]/
/// [`PostgresLog::consumer_lag`], so the dashboard and metrics code (which
/// know nothing about SQLite vs. PostgreSQL) can query either backend
/// through one object. Construct it by passing `Arc::new(your_log)` to
/// [`crate::RetainedLogLag::new`] for [`crate::Config::retained_log_lag`] -
/// blanket-implemented for both concrete log types below, not meant to be
/// implemented directly.
#[async_trait::async_trait]
pub trait LogLagSource: Send + Sync {
    /// Every consumer group's lag on every partition of `topic` that has
    /// committed at least once. A group that has never committed has no
    /// row here (its lag is unbounded/unknown - see `committed_offset`'s
    /// own "1 before first commit" default), rather than being reported as
    /// fully caught up.
    async fn consumer_lag(&self, topic: &str) -> anyhow::Result<Vec<ConsumerLag>>;
}

/// A fenced ownership claim for one consumer-group partition.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct LogPartitionClaim {
    /// Consumer group that owns the claim.
    pub group: String,
    /// Topic containing the claimed partition.
    pub topic: String,
    /// Claimed partition.
    pub partition: PartitionId,
    /// Live member that owns the lease.
    pub member_id: String,
    /// Fencing generation. A stale owner cannot commit after this changes.
    pub epoch: i64,
}

const MEMBER_TTL_SECONDS: i64 = 30;

fn validate_member_id(member_id: &str) -> anyhow::Result<()> {
    if member_id.trim().is_empty() {
        return Err(anyhow::anyhow!("consumer member ID cannot be empty"));
    }
    Ok(())
}

fn validate_read_limit(limit: u32) -> anyhow::Result<i64> {
    if limit == 0 || limit > 10_000 {
        return Err(anyhow::anyhow!("log read limit must be from 1 to 10,000"));
    }
    Ok(i64::from(limit))
}

fn validate_group_name(group: &str) -> anyhow::Result<()> {
    if group.trim().is_empty() {
        return Err(anyhow::anyhow!("consumer group name cannot be empty"));
    }
    Ok(())
}

fn validate_cleanup_offset(offset: i64) -> anyhow::Result<()> {
    if offset < 1 {
        return Err(anyhow::anyhow!("cleanup offset must be positive"));
    }
    Ok(())
}

#[cfg(feature = "sqlite")]
#[derive(Clone)]
/// A replayable log backed by SQLite.
///
/// Several processes on one host can share the same file and namespace.
pub struct SqliteLog {
    namespace: String,
    pool: sqlx::SqlitePool,
}

#[cfg(feature = "sqlite")]
impl SqliteLog {
    /// Creates a log using an application-owned migrated SQLite pool.
    pub async fn from_pool(
        namespace: impl Into<String>,
        pool: sqlx::SqlitePool,
    ) -> anyhow::Result<Self> {
        crate::storage::Sqlite::from_pool(pool.clone()).await?;
        Ok(Self {
            namespace: namespace.into(),
            pool,
        })
    }

    /// Opens SQLite and applies Later migrations.
    pub async fn connect(namespace: impl Into<String>, url: &str) -> anyhow::Result<Self> {
        let storage = crate::storage::Sqlite::new(url).await?;
        Ok(Self {
            namespace: namespace.into(),
            pool: storage.pool().clone(),
        })
    }

    /// Registers a fixed partition layout shared with sequential jobs.
    pub async fn register_topic(&self, topic: TopicConfig) -> anyhow::Result<()> {
        let count = i64::from(topic.partition_count());
        let result: Option<(i64,)> = sqlx::query_as(
            "INSERT INTO later_topic (namespace, topic, partition_count) VALUES (?1, ?2, ?3) \
             ON CONFLICT(namespace, topic) DO UPDATE SET partition_count = later_topic.partition_count \
             WHERE later_topic.partition_count = excluded.partition_count RETURNING partition_count",
        ).bind(&self.namespace).bind(topic.name()).bind(count).fetch_optional(&self.pool).await?;
        if result.is_none() {
            return Err(anyhow::anyhow!(
                "topic '{}' has a different partition count",
                topic.name()
            ));
        }
        Ok(())
    }

    /// Appends a record and returns its durable partition offset.
    pub async fn append(
        &self,
        topic: &str,
        key: PartitionKey,
        payload: Vec<u8>,
    ) -> anyhow::Result<LogRecord> {
        use sqlx::Row;
        let mut transaction = self.pool.begin().await?;
        let count: Option<(i64,)> = sqlx::query_as(
            "SELECT partition_count FROM later_topic WHERE namespace = ?1 AND topic = ?2",
        )
        .bind(&self.namespace)
        .bind(topic)
        .fetch_optional(&mut *transaction)
        .await?;
        let Some((count,)) = count else {
            return Err(anyhow::anyhow!("unknown topic '{topic}'"));
        };
        let partition = partition_for_key(&key, u32::try_from(count)?);
        let offset: i64 = sqlx::query("INSERT INTO later_log_partition (namespace, topic, partition_id, next_offset) VALUES (?1, ?2, ?3, 2) ON CONFLICT(namespace, topic, partition_id) DO UPDATE SET next_offset = later_log_partition.next_offset + 1 RETURNING next_offset - 1")
            .bind(&self.namespace).bind(topic).bind(i64::from(partition.0)).fetch_one(&mut *transaction).await?.try_get(0)?;
        let created_at =
            chrono::DateTime::from_timestamp_millis(chrono::Utc::now().timestamp_millis())
                .ok_or_else(|| anyhow::anyhow!("current timestamp is out of range"))?;
        sqlx::query("INSERT INTO later_log_record (namespace, topic, partition_id, record_offset, record_key, payload, created_at) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)")
            .bind(&self.namespace).bind(topic).bind(i64::from(partition.0)).bind(offset).bind(&key.0).bind(&payload).bind(created_at.timestamp_millis()).execute(&mut *transaction).await?;
        transaction.commit().await?;
        Ok(LogRecord {
            topic: topic.to_owned(),
            partition,
            offset,
            key,
            payload,
            created_at,
        })
    }

    /// Reads records at or after `offset`, in offset order.
    pub async fn read(
        &self,
        topic: &str,
        partition: PartitionId,
        offset: i64,
        limit: u32,
    ) -> anyhow::Result<Vec<LogRecord>> {
        use sqlx::Row;
        let limit = validate_read_limit(limit)?;
        let rows = sqlx::query("SELECT record_offset, record_key, payload, created_at FROM later_log_record WHERE namespace = ?1 AND topic = ?2 AND partition_id = ?3 AND record_offset >= ?4 ORDER BY record_offset LIMIT ?5")
            .bind(&self.namespace).bind(topic).bind(i64::from(partition.0)).bind(offset).bind(limit).fetch_all(&self.pool).await?;
        rows.into_iter()
            .map(|row| {
                let millis: i64 = row.try_get("created_at")?;
                let created_at = chrono::DateTime::from_timestamp_millis(millis)
                    .ok_or_else(|| anyhow::anyhow!("stored log timestamp is out of range"))?;
                Ok(LogRecord {
                    topic: topic.to_owned(),
                    partition,
                    offset: row.try_get("record_offset")?,
                    key: PartitionKey(row.try_get("record_key")?),
                    payload: row.try_get("payload")?,
                    created_at,
                })
            })
            .collect()
    }

    /// Stores the next offset a consumer group should read. Older commits are ignored.
    pub async fn commit_offset(
        &self,
        topic: &str,
        group: &str,
        partition: PartitionId,
        next_offset: i64,
    ) -> anyhow::Result<()> {
        validate_group_name(group)?;
        if next_offset < 1 {
            return Err(anyhow::anyhow!("next offset must be positive"));
        }
        sqlx::query("INSERT INTO later_log_consumer_offset (namespace, topic, group_name, partition_id, next_offset, committed_at) VALUES (?1, ?2, ?3, ?4, ?5, ?6) ON CONFLICT(namespace, topic, group_name, partition_id) DO UPDATE SET next_offset = excluded.next_offset, committed_at = excluded.committed_at WHERE excluded.next_offset > later_log_consumer_offset.next_offset")
            .bind(&self.namespace).bind(topic).bind(group).bind(i64::from(partition.0)).bind(next_offset).bind(chrono::Utc::now().timestamp_millis()).execute(&self.pool).await?;
        Ok(())
    }

    /// Returns the next offset for a group, or `1` before its first commit.
    pub async fn committed_offset(
        &self,
        topic: &str,
        group: &str,
        partition: PartitionId,
    ) -> anyhow::Result<i64> {
        validate_group_name(group)?;
        let row: Option<(i64,)> = sqlx::query_as("SELECT next_offset FROM later_log_consumer_offset WHERE namespace = ?1 AND topic = ?2 AND group_name = ?3 AND partition_id = ?4")
            .bind(&self.namespace).bind(topic).bind(group).bind(i64::from(partition.0)).fetch_optional(&self.pool).await?;
        Ok(row.map(|(offset,)| offset).unwrap_or(1))
    }

    /// Every consumer group's lag on every partition of `topic`. See
    /// [`ConsumerLag`] and [`LogLagSource::consumer_lag`].
    pub async fn consumer_lag(&self, topic: &str) -> anyhow::Result<Vec<ConsumerLag>> {
        use sqlx::Row;
        let rows = sqlx::query(
            "SELECT c.partition_id, c.group_name, c.next_offset AS committed_offset, \
                    p.next_offset AS end_offset \
             FROM later_log_consumer_offset c \
             JOIN later_log_partition p \
               ON p.namespace = c.namespace AND p.topic = c.topic AND p.partition_id = c.partition_id \
             WHERE c.namespace = ?1 AND c.topic = ?2",
        )
        .bind(&self.namespace)
        .bind(topic)
        .fetch_all(&self.pool)
        .await?;
        rows.into_iter()
            .map(|row| {
                let partition_id: i64 = row.try_get("partition_id")?;
                let committed_offset: i64 = row.try_get("committed_offset")?;
                let end_offset: i64 = row.try_get("end_offset")?;
                Ok(ConsumerLag {
                    partition: PartitionId(u32::try_from(partition_id)?),
                    group: row.try_get("group_name")?,
                    end_offset,
                    committed_offset,
                    lag: (end_offset - committed_offset).max(0),
                })
            })
            .collect()
    }

    /// Returns the first offset that every registered group has passed.
    pub async fn safe_cleanup_offset(
        &self,
        topic: &str,
        partition: PartitionId,
    ) -> anyhow::Result<Option<i64>> {
        let row: (Option<i64>,) = sqlx::query_as("SELECT MIN(next_offset) FROM later_log_consumer_offset WHERE namespace = ?1 AND topic = ?2 AND partition_id = ?3")
            .bind(&self.namespace).bind(topic).bind(i64::from(partition.0)).fetch_one(&self.pool).await?;
        Ok(row.0)
    }

    /// Deletes records before `offset` only when all groups have committed past it.
    ///
    /// The safety bound is recomputed live inside the `DELETE` itself, not
    /// from the earlier [`Self::safe_cleanup_offset`] read below used only
    /// to validate `offset` up front: a group's very first commit landing
    /// between that read and the delete must still protect its unread
    /// records, not be deleted out from under it because the read it raced
    /// didn't know it existed yet.
    pub async fn delete_before(
        &self,
        topic: &str,
        partition: PartitionId,
        offset: i64,
    ) -> anyhow::Result<u64> {
        validate_cleanup_offset(offset)?;
        let Some(safe_offset) = self.safe_cleanup_offset(topic, partition).await? else {
            return Ok(0);
        };
        if offset > safe_offset {
            return Err(anyhow::anyhow!(
                "cleanup offset is ahead of an active consumer group"
            ));
        }
        Ok(sqlx::query(
            "DELETE FROM later_log_record WHERE namespace = ?1 AND topic = ?2 AND partition_id = ?3 \
             AND record_offset < ?4 \
             AND record_offset < COALESCE( \
                 (SELECT MIN(next_offset) FROM later_log_consumer_offset \
                  WHERE namespace = ?1 AND topic = ?2 AND partition_id = ?3), 0)",
        )
        .bind(&self.namespace).bind(topic).bind(i64::from(partition.0)).bind(offset).execute(&self.pool).await?.rows_affected())
    }

    /// Removes superseded keys below every group's committed offset.
    ///
    /// Same reasoning as [`Self::delete_before`]: the safety bound is
    /// recomputed live inside the `DELETE`, not carried over from an
    /// earlier read, so a group's first-ever commit racing this call is
    /// still honored.
    pub async fn compact_partition(
        &self,
        topic: &str,
        partition: PartitionId,
    ) -> anyhow::Result<u64> {
        Ok(sqlx::query(
            "DELETE FROM later_log_record AS older WHERE older.namespace = ?1 AND older.topic = ?2 AND older.partition_id = ?3 \
             AND older.record_offset < COALESCE( \
                 (SELECT MIN(next_offset) FROM later_log_consumer_offset \
                  WHERE namespace = ?1 AND topic = ?2 AND partition_id = ?3), 0) \
             AND EXISTS (SELECT 1 FROM later_log_record AS newer WHERE newer.namespace = older.namespace AND newer.topic = older.topic AND newer.partition_id = older.partition_id AND newer.record_key = older.record_key AND newer.record_offset > older.record_offset)",
        )
        .bind(&self.namespace).bind(topic).bind(i64::from(partition.0)).execute(&self.pool).await?.rows_affected())
    }

    /// Marks a consumer member live for the next 30 seconds.
    pub async fn heartbeat(&self, topic: &str, group: &str, member_id: &str) -> anyhow::Result<()> {
        validate_group_name(group)?;
        validate_member_id(member_id)?;
        sqlx::query("INSERT INTO later_log_consumer_member (namespace, topic, group_name, member_id, heartbeat_at) VALUES (?1, ?2, ?3, ?4, ?5) ON CONFLICT(namespace, topic, group_name, member_id) DO UPDATE SET heartbeat_at = excluded.heartbeat_at")
            .bind(&self.namespace).bind(topic).bind(group).bind(member_id).bind(chrono::Utc::now().timestamp_millis()).execute(&self.pool).await?;
        Ok(())
    }

    /// Claims this member's rendezvous-assigned partition until its lease expires.
    pub async fn claim_partition(
        &self,
        topic: &str,
        group: &str,
        member_id: &str,
        partition: PartitionId,
    ) -> anyhow::Result<Option<LogPartitionClaim>> {
        use sqlx::Row;
        self.heartbeat(topic, group, member_id).await?;
        let now = chrono::Utc::now().timestamp_millis();
        let members: Vec<(String,)> = sqlx::query_as("SELECT member_id FROM later_log_consumer_member WHERE namespace = ?1 AND topic = ?2 AND group_name = ?3 AND heartbeat_at > ?4")
            .bind(&self.namespace).bind(topic).bind(group).bind(now - MEMBER_TTL_SECONDS * 1000).fetch_all(&self.pool).await?;
        let members = members
            .into_iter()
            .map(|(member,)| member)
            .collect::<Vec<_>>();
        if crate::topic::rendezvous_owner(topic, partition, &members) != Some(member_id) {
            return Ok(None);
        }
        let lease_until = now + MEMBER_TTL_SECONDS * 1000;
        let row = sqlx::query("INSERT INTO later_log_consumer_lease (namespace, topic, group_name, partition_id, member_id, lease_until, lease_epoch) VALUES (?1, ?2, ?3, ?4, ?5, ?6, 1) ON CONFLICT(namespace, topic, group_name, partition_id) DO UPDATE SET member_id = excluded.member_id, lease_until = excluded.lease_until, lease_epoch = later_log_consumer_lease.lease_epoch + 1 WHERE later_log_consumer_lease.lease_until <= ?7 RETURNING lease_epoch")
            .bind(&self.namespace).bind(topic).bind(group).bind(i64::from(partition.0)).bind(member_id).bind(lease_until).bind(now).fetch_optional(&self.pool).await?;
        let Some(row) = row else {
            return Ok(None);
        };
        Ok(Some(LogPartitionClaim {
            group: group.to_owned(),
            topic: topic.to_owned(),
            partition,
            member_id: member_id.to_owned(),
            epoch: row.try_get(0)?,
        }))
    }

    /// Renews a claim. Returns `false` after the lease was lost or fenced.
    pub async fn renew_claim(&self, claim: &LogPartitionClaim) -> anyhow::Result<bool> {
        let now = chrono::Utc::now().timestamp_millis();
        let result = sqlx::query("UPDATE later_log_consumer_lease SET lease_until = ?1 WHERE namespace = ?2 AND topic = ?3 AND group_name = ?4 AND partition_id = ?5 AND member_id = ?6 AND lease_epoch = ?7 AND lease_until > ?8")
            .bind(now + MEMBER_TTL_SECONDS * 1000).bind(&self.namespace).bind(&claim.topic).bind(&claim.group).bind(i64::from(claim.partition.0)).bind(&claim.member_id).bind(claim.epoch).bind(now).execute(&self.pool).await?;
        Ok(result.rows_affected() == 1)
    }

    /// Releases a claim when its current member is done with the partition.
    pub async fn release_claim(&self, claim: &LogPartitionClaim) -> anyhow::Result<bool> {
        let result = sqlx::query("DELETE FROM later_log_consumer_lease WHERE namespace = ?1 AND topic = ?2 AND group_name = ?3 AND partition_id = ?4 AND member_id = ?5 AND lease_epoch = ?6")
            .bind(&self.namespace).bind(&claim.topic).bind(&claim.group).bind(i64::from(claim.partition.0)).bind(&claim.member_id).bind(claim.epoch).execute(&self.pool).await?;
        Ok(result.rows_affected() == 1)
    }

    /// Removes a member from a group and releases every lease it still owns.
    pub async fn leave_group(
        &self,
        topic: &str,
        group: &str,
        member_id: &str,
    ) -> anyhow::Result<()> {
        validate_group_name(group)?;
        validate_member_id(member_id)?;
        let mut transaction = self.pool.begin().await?;
        sqlx::query("DELETE FROM later_log_consumer_member WHERE namespace = ?1 AND topic = ?2 AND group_name = ?3 AND member_id = ?4")
            .bind(&self.namespace).bind(topic).bind(group).bind(member_id).execute(&mut *transaction).await?;
        sqlx::query("DELETE FROM later_log_consumer_lease WHERE namespace = ?1 AND topic = ?2 AND group_name = ?3 AND member_id = ?4")
            .bind(&self.namespace).bind(topic).bind(group).bind(member_id).execute(&mut *transaction).await?;
        transaction.commit().await?;
        Ok(())
    }

    /// Commits an offset only while `claim` remains the current fenced owner.
    pub async fn commit_claimed_offset(
        &self,
        claim: &LogPartitionClaim,
        next_offset: i64,
    ) -> anyhow::Result<bool> {
        if next_offset < 1 {
            return Err(anyhow::anyhow!("next offset must be positive"));
        }
        let now = chrono::Utc::now().timestamp_millis();
        let mut transaction = self.pool.begin().await?;
        let owned: Option<(i64,)> = sqlx::query_as("SELECT lease_epoch FROM later_log_consumer_lease WHERE namespace = ?1 AND topic = ?2 AND group_name = ?3 AND partition_id = ?4 AND member_id = ?5 AND lease_epoch = ?6 AND lease_until > ?7")
            .bind(&self.namespace).bind(&claim.topic).bind(&claim.group).bind(i64::from(claim.partition.0)).bind(&claim.member_id).bind(claim.epoch).bind(now).fetch_optional(&mut *transaction).await?;
        if owned.is_none() {
            transaction.rollback().await?;
            return Ok(false);
        }
        sqlx::query("INSERT INTO later_log_consumer_offset (namespace, topic, group_name, partition_id, next_offset, committed_at) VALUES (?1, ?2, ?3, ?4, ?5, ?6) ON CONFLICT(namespace, topic, group_name, partition_id) DO UPDATE SET next_offset = excluded.next_offset, committed_at = excluded.committed_at WHERE excluded.next_offset > later_log_consumer_offset.next_offset")
            .bind(&self.namespace).bind(&claim.topic).bind(&claim.group).bind(i64::from(claim.partition.0)).bind(next_offset).bind(now).execute(&mut *transaction).await?;
        transaction.commit().await?;
        Ok(true)
    }
}

#[cfg(feature = "postgres")]
#[derive(Clone)]
/// A replayable log backed by PostgreSQL.
///
/// Servers on many hosts can share a database and namespace.
pub struct PostgresLog {
    namespace: String,
    pool: sqlx::PgPool,
}

#[cfg(feature = "postgres")]
impl PostgresLog {
    /// Creates a log using an application-owned migrated PostgreSQL pool.
    pub async fn from_pool(
        namespace: impl Into<String>,
        pool: sqlx::PgPool,
    ) -> anyhow::Result<Self> {
        crate::storage::Postgres::from_pool(pool.clone()).await?;
        Ok(Self {
            namespace: namespace.into(),
            pool,
        })
    }

    /// Connects to PostgreSQL and applies Later migrations.
    pub async fn connect(namespace: impl Into<String>, url: &str) -> anyhow::Result<Self> {
        let storage = crate::storage::Postgres::new(url).await?;
        Ok(Self {
            namespace: namespace.into(),
            pool: storage.pool().clone(),
        })
    }

    /// Registers a fixed partition layout shared with sequential jobs.
    pub async fn register_topic(&self, topic: TopicConfig) -> anyhow::Result<()> {
        let count = i64::from(topic.partition_count());
        let result: Option<(i64,)> = sqlx::query_as(
            "INSERT INTO later_topic (namespace, topic, partition_count) VALUES ($1, $2, $3) \
             ON CONFLICT(namespace, topic) DO UPDATE SET partition_count = later_topic.partition_count \
             WHERE later_topic.partition_count = excluded.partition_count RETURNING partition_count",
        ).bind(&self.namespace).bind(topic.name()).bind(count).fetch_optional(&self.pool).await?;
        if result.is_none() {
            return Err(anyhow::anyhow!(
                "topic '{}' has a different partition count",
                topic.name()
            ));
        }
        Ok(())
    }

    /// Appends a record and returns its durable partition offset.
    pub async fn append(
        &self,
        topic: &str,
        key: PartitionKey,
        payload: Vec<u8>,
    ) -> anyhow::Result<LogRecord> {
        use sqlx::Row;
        let mut transaction = self.pool.begin().await?;
        let count: Option<(i64,)> = sqlx::query_as(
            "SELECT partition_count FROM later_topic WHERE namespace = $1 AND topic = $2",
        )
        .bind(&self.namespace)
        .bind(topic)
        .fetch_optional(&mut *transaction)
        .await?;
        let Some((count,)) = count else {
            return Err(anyhow::anyhow!("unknown topic '{topic}'"));
        };
        let partition = partition_for_key(&key, u32::try_from(count)?);
        let offset: i64 = sqlx::query("INSERT INTO later_log_partition (namespace, topic, partition_id, next_offset) VALUES ($1, $2, $3, 2) ON CONFLICT(namespace, topic, partition_id) DO UPDATE SET next_offset = later_log_partition.next_offset + 1 RETURNING next_offset - 1")
            .bind(&self.namespace).bind(topic).bind(i64::from(partition.0)).fetch_one(&mut *transaction).await?.try_get(0)?;
        // Bind `NOW()` and read back whatever Postgres actually stored
        // (TIMESTAMPTZ only keeps microsecond precision) rather than the
        // nanosecond-precision `chrono::Utc::now()` a client could compute,
        // so a freshly appended record's `created_at` always matches what a
        // later `read()` of the same row returns.
        let created_at: crate::UtcDateTime = sqlx::query("INSERT INTO later_log_record (namespace, topic, partition_id, record_offset, record_key, payload, created_at) VALUES ($1, $2, $3, $4, $5, $6, NOW()) RETURNING created_at")
            .bind(&self.namespace).bind(topic).bind(i64::from(partition.0)).bind(offset).bind(&key.0).bind(&payload).fetch_one(&mut *transaction).await?.try_get("created_at")?;
        transaction.commit().await?;
        Ok(LogRecord {
            topic: topic.to_owned(),
            partition,
            offset,
            key,
            payload,
            created_at,
        })
    }

    /// Reads records at or after `offset`, in offset order.
    pub async fn read(
        &self,
        topic: &str,
        partition: PartitionId,
        offset: i64,
        limit: u32,
    ) -> anyhow::Result<Vec<LogRecord>> {
        use sqlx::Row;
        let limit = validate_read_limit(limit)?;
        let rows = sqlx::query("SELECT record_offset, record_key, payload, created_at FROM later_log_record WHERE namespace = $1 AND topic = $2 AND partition_id = $3 AND record_offset >= $4 ORDER BY record_offset LIMIT $5")
            .bind(&self.namespace).bind(topic).bind(i64::from(partition.0)).bind(offset).bind(limit).fetch_all(&self.pool).await?;
        rows.into_iter()
            .map(|row| {
                Ok(LogRecord {
                    topic: topic.to_owned(),
                    partition,
                    offset: row.try_get("record_offset")?,
                    key: PartitionKey(row.try_get("record_key")?),
                    payload: row.try_get("payload")?,
                    created_at: row.try_get("created_at")?,
                })
            })
            .collect()
    }

    /// Stores the next offset a consumer group should read. Older commits are ignored.
    pub async fn commit_offset(
        &self,
        topic: &str,
        group: &str,
        partition: PartitionId,
        next_offset: i64,
    ) -> anyhow::Result<()> {
        validate_group_name(group)?;
        if next_offset < 1 {
            return Err(anyhow::anyhow!("next offset must be positive"));
        }
        sqlx::query("INSERT INTO later_log_consumer_offset (namespace, topic, group_name, partition_id, next_offset) VALUES ($1, $2, $3, $4, $5) ON CONFLICT(namespace, topic, group_name, partition_id) DO UPDATE SET next_offset = excluded.next_offset, committed_at = NOW() WHERE excluded.next_offset > later_log_consumer_offset.next_offset")
            .bind(&self.namespace).bind(topic).bind(group).bind(i64::from(partition.0)).bind(next_offset).execute(&self.pool).await?;
        Ok(())
    }

    /// Returns the next offset for a group, or `1` before its first commit.
    pub async fn committed_offset(
        &self,
        topic: &str,
        group: &str,
        partition: PartitionId,
    ) -> anyhow::Result<i64> {
        validate_group_name(group)?;
        let row: Option<(i64,)> = sqlx::query_as("SELECT next_offset FROM later_log_consumer_offset WHERE namespace = $1 AND topic = $2 AND group_name = $3 AND partition_id = $4")
            .bind(&self.namespace).bind(topic).bind(group).bind(i64::from(partition.0)).fetch_optional(&self.pool).await?;
        Ok(row.map(|(offset,)| offset).unwrap_or(1))
    }

    /// Every consumer group's lag on every partition of `topic`. See
    /// [`ConsumerLag`] and [`LogLagSource::consumer_lag`].
    pub async fn consumer_lag(&self, topic: &str) -> anyhow::Result<Vec<ConsumerLag>> {
        use sqlx::Row;
        let rows = sqlx::query(
            "SELECT c.partition_id, c.group_name, c.next_offset AS committed_offset, \
                    p.next_offset AS end_offset \
             FROM later_log_consumer_offset c \
             JOIN later_log_partition p \
               ON p.namespace = c.namespace AND p.topic = c.topic AND p.partition_id = c.partition_id \
             WHERE c.namespace = $1 AND c.topic = $2",
        )
        .bind(&self.namespace)
        .bind(topic)
        .fetch_all(&self.pool)
        .await?;
        rows.into_iter()
            .map(|row| {
                let partition_id: i64 = row.try_get("partition_id")?;
                let committed_offset: i64 = row.try_get("committed_offset")?;
                let end_offset: i64 = row.try_get("end_offset")?;
                Ok(ConsumerLag {
                    partition: PartitionId(u32::try_from(partition_id)?),
                    group: row.try_get("group_name")?,
                    end_offset,
                    committed_offset,
                    lag: (end_offset - committed_offset).max(0),
                })
            })
            .collect()
    }

    /// Returns the first offset that every registered group has passed.
    pub async fn safe_cleanup_offset(
        &self,
        topic: &str,
        partition: PartitionId,
    ) -> anyhow::Result<Option<i64>> {
        let row: (Option<i64>,) = sqlx::query_as("SELECT MIN(next_offset) FROM later_log_consumer_offset WHERE namespace = $1 AND topic = $2 AND partition_id = $3")
            .bind(&self.namespace).bind(topic).bind(i64::from(partition.0)).fetch_one(&self.pool).await?;
        Ok(row.0)
    }

    /// Deletes records before `offset` only when all groups have committed
    /// past it.
    ///
    /// The safety bound is recomputed live inside the `DELETE` itself, not
    /// from the earlier [`Self::safe_cleanup_offset`] read below used only
    /// to validate `offset` up front: a group's very first commit landing
    /// between that read and the delete must still protect its unread
    /// records, not be deleted out from under it because the read it raced
    /// didn't know it existed yet.
    pub async fn delete_before(
        &self,
        topic: &str,
        partition: PartitionId,
        offset: i64,
    ) -> anyhow::Result<u64> {
        validate_cleanup_offset(offset)?;
        let Some(safe_offset) = self.safe_cleanup_offset(topic, partition).await? else {
            return Ok(0);
        };
        if offset > safe_offset {
            return Err(anyhow::anyhow!(
                "cleanup offset is ahead of an active consumer group"
            ));
        }
        Ok(sqlx::query(
            "DELETE FROM later_log_record WHERE namespace = $1 AND topic = $2 AND partition_id = $3 \
             AND record_offset < $4 \
             AND record_offset < COALESCE( \
                 (SELECT MIN(next_offset) FROM later_log_consumer_offset \
                  WHERE namespace = $1 AND topic = $2 AND partition_id = $3), 0)",
        )
        .bind(&self.namespace).bind(topic).bind(i64::from(partition.0)).bind(offset).execute(&self.pool).await?.rows_affected())
    }

    /// Removes superseded keys below every group's committed offset.
    ///
    /// Same reasoning as [`Self::delete_before`]: the safety bound is
    /// recomputed live inside the `DELETE`, not carried over from an
    /// earlier read, so a group's first-ever commit racing this call is
    /// still honored.
    pub async fn compact_partition(
        &self,
        topic: &str,
        partition: PartitionId,
    ) -> anyhow::Result<u64> {
        Ok(sqlx::query(
            "DELETE FROM later_log_record AS older WHERE older.namespace = $1 AND older.topic = $2 AND older.partition_id = $3 \
             AND older.record_offset < COALESCE( \
                 (SELECT MIN(next_offset) FROM later_log_consumer_offset \
                  WHERE namespace = $1 AND topic = $2 AND partition_id = $3), 0) \
             AND EXISTS (SELECT 1 FROM later_log_record AS newer WHERE newer.namespace = older.namespace AND newer.topic = older.topic AND newer.partition_id = older.partition_id AND newer.record_key = older.record_key AND newer.record_offset > older.record_offset)",
        )
        .bind(&self.namespace).bind(topic).bind(i64::from(partition.0)).execute(&self.pool).await?.rows_affected())
    }

    /// Marks a consumer member live for the next 30 seconds.
    pub async fn heartbeat(&self, topic: &str, group: &str, member_id: &str) -> anyhow::Result<()> {
        validate_group_name(group)?;
        validate_member_id(member_id)?;
        sqlx::query("INSERT INTO later_log_consumer_member (namespace, topic, group_name, member_id) VALUES ($1, $2, $3, $4) ON CONFLICT(namespace, topic, group_name, member_id) DO UPDATE SET heartbeat_at = NOW()")
            .bind(&self.namespace).bind(topic).bind(group).bind(member_id).execute(&self.pool).await?;
        Ok(())
    }

    /// Claims this member's rendezvous-assigned partition until its lease expires.
    pub async fn claim_partition(
        &self,
        topic: &str,
        group: &str,
        member_id: &str,
        partition: PartitionId,
    ) -> anyhow::Result<Option<LogPartitionClaim>> {
        use sqlx::Row;
        self.heartbeat(topic, group, member_id).await?;
        let members: Vec<(String,)> = sqlx::query_as("SELECT member_id FROM later_log_consumer_member WHERE namespace = $1 AND topic = $2 AND group_name = $3 AND heartbeat_at > NOW() - INTERVAL '30 seconds'")
            .bind(&self.namespace).bind(topic).bind(group).fetch_all(&self.pool).await?;
        let members = members
            .into_iter()
            .map(|(member,)| member)
            .collect::<Vec<_>>();
        if crate::topic::rendezvous_owner(topic, partition, &members) != Some(member_id) {
            return Ok(None);
        }
        let row = sqlx::query("INSERT INTO later_log_consumer_lease (namespace, topic, group_name, partition_id, member_id, lease_until, lease_epoch) VALUES ($1, $2, $3, $4, $5, NOW() + INTERVAL '30 seconds', 1) ON CONFLICT(namespace, topic, group_name, partition_id) DO UPDATE SET member_id = excluded.member_id, lease_until = excluded.lease_until, lease_epoch = later_log_consumer_lease.lease_epoch + 1 WHERE later_log_consumer_lease.lease_until <= NOW() RETURNING lease_epoch")
            .bind(&self.namespace).bind(topic).bind(group).bind(i64::from(partition.0)).bind(member_id).fetch_optional(&self.pool).await?;
        let Some(row) = row else {
            return Ok(None);
        };
        Ok(Some(LogPartitionClaim {
            group: group.to_owned(),
            topic: topic.to_owned(),
            partition,
            member_id: member_id.to_owned(),
            epoch: row.try_get(0)?,
        }))
    }
    /// Renews a claim. Returns `false` after the lease was lost or fenced.
    pub async fn renew_claim(&self, claim: &LogPartitionClaim) -> anyhow::Result<bool> {
        let result = sqlx::query("UPDATE later_log_consumer_lease SET lease_until = NOW() + INTERVAL '30 seconds' WHERE namespace = $1 AND topic = $2 AND group_name = $3 AND partition_id = $4 AND member_id = $5 AND lease_epoch = $6 AND lease_until > NOW()")
            .bind(&self.namespace).bind(&claim.topic).bind(&claim.group).bind(i64::from(claim.partition.0)).bind(&claim.member_id).bind(claim.epoch).execute(&self.pool).await?;
        Ok(result.rows_affected() == 1)
    }

    /// Releases a claim when its current member is done with the partition.
    pub async fn release_claim(&self, claim: &LogPartitionClaim) -> anyhow::Result<bool> {
        let result = sqlx::query("DELETE FROM later_log_consumer_lease WHERE namespace = $1 AND topic = $2 AND group_name = $3 AND partition_id = $4 AND member_id = $5 AND lease_epoch = $6")
            .bind(&self.namespace).bind(&claim.topic).bind(&claim.group).bind(i64::from(claim.partition.0)).bind(&claim.member_id).bind(claim.epoch).execute(&self.pool).await?;
        Ok(result.rows_affected() == 1)
    }

    /// Removes a member from a group and releases every lease it still owns.
    pub async fn leave_group(
        &self,
        topic: &str,
        group: &str,
        member_id: &str,
    ) -> anyhow::Result<()> {
        validate_group_name(group)?;
        validate_member_id(member_id)?;
        let mut transaction = self.pool.begin().await?;
        sqlx::query("DELETE FROM later_log_consumer_member WHERE namespace = $1 AND topic = $2 AND group_name = $3 AND member_id = $4")
            .bind(&self.namespace).bind(topic).bind(group).bind(member_id).execute(&mut *transaction).await?;
        sqlx::query("DELETE FROM later_log_consumer_lease WHERE namespace = $1 AND topic = $2 AND group_name = $3 AND member_id = $4")
            .bind(&self.namespace).bind(topic).bind(group).bind(member_id).execute(&mut *transaction).await?;
        transaction.commit().await?;
        Ok(())
    }

    /// Commits an offset only while `claim` remains the current fenced owner.
    pub async fn commit_claimed_offset(
        &self,
        claim: &LogPartitionClaim,
        next_offset: i64,
    ) -> anyhow::Result<bool> {
        if next_offset < 1 {
            return Err(anyhow::anyhow!("next offset must be positive"));
        }
        let mut transaction = self.pool.begin().await?;
        let owned: Option<(i64,)> = sqlx::query_as("SELECT lease_epoch FROM later_log_consumer_lease WHERE namespace = $1 AND topic = $2 AND group_name = $3 AND partition_id = $4 AND member_id = $5 AND lease_epoch = $6 AND lease_until > NOW()")
            .bind(&self.namespace).bind(&claim.topic).bind(&claim.group).bind(i64::from(claim.partition.0)).bind(&claim.member_id).bind(claim.epoch).fetch_optional(&mut *transaction).await?;
        if owned.is_none() {
            transaction.rollback().await?;
            return Ok(false);
        }
        sqlx::query("INSERT INTO later_log_consumer_offset (namespace, topic, group_name, partition_id, next_offset) VALUES ($1, $2, $3, $4, $5) ON CONFLICT(namespace, topic, group_name, partition_id) DO UPDATE SET next_offset = excluded.next_offset, committed_at = NOW() WHERE excluded.next_offset > later_log_consumer_offset.next_offset")
            .bind(&self.namespace).bind(&claim.topic).bind(&claim.group).bind(i64::from(claim.partition.0)).bind(next_offset).execute(&mut *transaction).await?;
        transaction.commit().await?;
        Ok(true)
    }
}

#[cfg(feature = "sqlite")]
#[async_trait::async_trait]
impl LogLagSource for SqliteLog {
    async fn consumer_lag(&self, topic: &str) -> anyhow::Result<Vec<ConsumerLag>> {
        self.consumer_lag(topic).await
    }
}

#[cfg(feature = "postgres")]
#[async_trait::async_trait]
impl LogLagSource for PostgresLog {
    async fn consumer_lag(&self, topic: &str) -> anyhow::Result<Vec<ConsumerLag>> {
        self.consumer_lag(topic).await
    }
}

#[cfg(all(test, feature = "sqlite"))]
mod sqlite_tests {
    use super::*;

    /// `delete_before`/`compact_partition` recompute their safety bound
    /// live inside the `DELETE` itself now, instead of from an earlier,
    /// separate `safe_cleanup_offset` read - this pins down that with zero
    /// registered consumer groups (`MIN(next_offset)` is `NULL`), the
    /// `COALESCE(..., 0)` fallback still means "delete nothing", the same
    /// as the old early-return did.
    #[tokio::test]
    async fn cleanup_with_no_consumer_groups_deletes_nothing() -> anyhow::Result<()> {
        let log = SqliteLog::connect("log-test", "sqlite::memory:").await?;
        log.register_topic(TopicConfig::new("orphan", 1)?).await?;
        log.append("orphan", PartitionKey::from("a"), b"1".to_vec())
            .await?;
        log.append("orphan", PartitionKey::from("a"), b"2".to_vec())
            .await?;

        assert_eq!(
            log.safe_cleanup_offset("orphan", PartitionId(0)).await?,
            None
        );
        assert_eq!(log.compact_partition("orphan", PartitionId(0)).await?, 0);
        assert_eq!(
            log.read("orphan", PartitionId(0), 1, 10).await?.len(),
            2,
            "compact_partition must not have deleted anything with no consumer groups"
        );
        Ok(())
    }

    /// A normal (append-only) topic: every record has a distinct key, so
    /// `compact_partition` never removes anything. Only `delete_before`
    /// trims history, and only once every registered consumer group has
    /// committed past the requested offset.
    #[tokio::test]
    async fn normal_topic_retains_every_record_until_explicitly_trimmed() -> anyhow::Result<()> {
        let log = SqliteLog::connect("log-test", "sqlite::memory:").await?;
        log.register_topic(TopicConfig::new("events", 1)?).await?;

        let mut records = Vec::new();
        for index in 1..=5 {
            records.push(
                log.append(
                    "events",
                    PartitionKey::from(format!("event-{index}")),
                    format!("payload-{index}").into_bytes(),
                )
                .await?,
            );
        }
        assert_eq!(
            records.iter().map(|r| r.offset).collect::<Vec<_>>(),
            vec![1, 2, 3, 4, 5]
        );

        // Two independent consumer groups at different points.
        log.commit_offset("events", "audit", PartitionId(0), 6)
            .await?;
        log.commit_offset("events", "search", PartitionId(0), 3)
            .await?;

        // No key is ever repeated, so compaction is a no-op regardless of
        // how far consumers have advanced.
        assert_eq!(log.compact_partition("events", PartitionId(0)).await?, 0);
        assert_eq!(
            log.read("events", PartitionId(0), 1, 10).await?,
            records.clone()
        );

        // The safe watermark is the *slowest* group ("search" at 3), so
        // trimming past it is rejected even though "audit" is fully caught up.
        assert_eq!(
            log.safe_cleanup_offset("events", PartitionId(0)).await?,
            Some(3)
        );
        assert!(log
            .delete_before("events", PartitionId(0), 4)
            .await
            .is_err());

        // Trimming up to the safe watermark removes exactly the consumed prefix.
        assert_eq!(log.delete_before("events", PartitionId(0), 3).await?, 2);
        assert_eq!(
            log.read("events", PartitionId(0), 1, 10).await?,
            records[2..].to_vec()
        );

        // Once the slow group catches up, the rest can be trimmed too.
        log.commit_offset("events", "search", PartitionId(0), 6)
            .await?;
        assert_eq!(log.delete_before("events", PartitionId(0), 6).await?, 3);
        assert_eq!(log.read("events", PartitionId(0), 1, 10).await?, Vec::new());
        Ok(())
    }

    /// `consumer_lag` reports each group's own distance from the end of the
    /// partition, not from each other - unlike `safe_cleanup_offset`, a
    /// fast group's reading is unaffected by a slow one's.
    #[tokio::test]
    async fn consumer_lag_reports_each_groups_own_distance_from_the_end() -> anyhow::Result<()> {
        let log = SqliteLog::connect("log-test", "sqlite::memory:").await?;
        log.register_topic(TopicConfig::new("events", 1)?).await?;

        // No commits yet: nothing to report (a group that has never
        // committed has no lag reading - see `committed_offset`'s own "1
        // before first commit" default).
        assert_eq!(log.consumer_lag("events").await?, Vec::new());

        for index in 1..=5 {
            log.append(
                "events",
                PartitionKey::from(format!("event-{index}")),
                format!("payload-{index}").into_bytes(),
            )
            .await?;
        }
        // End offset is now 6 (5 records written, next append would be 6).
        log.commit_offset("events", "caught-up", PartitionId(0), 6)
            .await?;
        log.commit_offset("events", "behind", PartitionId(0), 3)
            .await?;

        let mut lag = log.consumer_lag("events").await?;
        lag.sort_by(|a, b| a.group.cmp(&b.group));
        assert_eq!(
            lag,
            vec![
                ConsumerLag {
                    partition: PartitionId(0),
                    group: "behind".to_string(),
                    end_offset: 6,
                    committed_offset: 3,
                    lag: 3,
                },
                ConsumerLag {
                    partition: PartitionId(0),
                    group: "caught-up".to_string(),
                    end_offset: 6,
                    committed_offset: 6,
                    lag: 0,
                },
            ]
        );
        Ok(())
    }

    /// A compacted topic: records reuse a key per logical entity.
    /// `compact_partition` keeps only the latest record per key, never
    /// touches a key written only once, and never runs ahead of the
    /// slowest registered consumer group.
    #[tokio::test]
    async fn compacted_topic_keeps_only_the_latest_record_per_key() -> anyhow::Result<()> {
        let log = SqliteLog::connect("log-test", "sqlite::memory:").await?;
        log.register_topic(TopicConfig::new("orders", 1)?).await?;

        let order_one = PartitionKey::from("order-1");
        let order_two = PartitionKey::from("order-2");
        let order_three = PartitionKey::from("order-3");

        log.append("orders", order_one.clone(), b"v1".to_vec())
            .await?; // offset 1, superseded
        log.append("orders", order_two.clone(), b"v1".to_vec())
            .await?; // offset 2, superseded
        log.append("orders", order_one.clone(), b"v2".to_vec())
            .await?; // offset 3, superseded
        log.append("orders", order_three.clone(), b"only".to_vec())
            .await?; // offset 4, never superseded
        let order_one_latest = log
            .append("orders", order_one.clone(), b"v3".to_vec())
            .await?; // offset 5, latest for order-1
        let order_two_latest = log
            .append("orders", order_two.clone(), b"v2".to_vec())
            .await?; // offset 6, latest for order-2

        // A fully caught-up projector and a slow reader stuck at offset 4.
        log.commit_offset("orders", "projector", PartitionId(0), 7)
            .await?;
        log.commit_offset("orders", "slow-reader", PartitionId(0), 4)
            .await?;
        assert_eq!(
            log.safe_cleanup_offset("orders", PartitionId(0)).await?,
            Some(4)
        );

        // Only offsets 1-3 are below the watermark; of those, 1 and 3 are
        // superseded copies of order-1 and 2 is a superseded copy of
        // order-2. The still-unread offset 4 (order-3, written once) is
        // untouched either way.
        assert_eq!(log.compact_partition("orders", PartitionId(0)).await?, 3);
        let order_three_record = log
            .read("orders", PartitionId(0), 1, 1)
            .await?
            .into_iter()
            .next()
            .ok_or_else(|| anyhow::anyhow!("order-3's only record should survive compaction"))?;
        assert_eq!(order_three_record.key, order_three);
        assert_eq!(order_three_record.payload, b"only");

        // Compaction is idempotent: nothing left below the watermark to remove.
        assert_eq!(log.compact_partition("orders", PartitionId(0)).await?, 0);

        // The slow reader catches up; the watermark now covers everything,
        // but the surviving records are already exactly the latest per key.
        log.commit_offset("orders", "slow-reader", PartitionId(0), 7)
            .await?;
        assert_eq!(log.compact_partition("orders", PartitionId(0)).await?, 0);
        assert_eq!(
            log.read("orders", PartitionId(0), 1, 10).await?,
            vec![order_three_record, order_one_latest, order_two_latest]
        );
        Ok(())
    }

    #[tokio::test]
    async fn append_and_replay_keep_partition_offsets() -> anyhow::Result<()> {
        let log = SqliteLog::connect("log-test", "sqlite::memory:").await?;
        log.register_topic(TopicConfig::new("orders", 1)?).await?;
        let first = log
            .append("orders", PartitionKey::from("one"), b"first".to_vec())
            .await?;
        let second = log
            .append("orders", PartitionKey::from("two"), b"second".to_vec())
            .await?;
        assert_eq!((first.offset, second.offset), (1, 2));
        assert_eq!(
            log.read("orders", PartitionId(0), 1, 10).await?,
            vec![first.clone(), second.clone()]
        );
        assert_eq!(
            log.committed_offset("orders", "billing", PartitionId(0))
                .await?,
            1
        );
        log.commit_offset("orders", "billing", PartitionId(0), 3)
            .await?;
        log.commit_offset("orders", "billing", PartitionId(0), 2)
            .await?;
        assert_eq!(
            log.committed_offset("orders", "billing", PartitionId(0))
                .await?,
            3
        );
        let replacement = log
            .append("orders", PartitionKey::from("one"), b"replacement".to_vec())
            .await?;
        log.commit_offset("orders", "billing", PartitionId(0), 4)
            .await?;
        assert_eq!(log.compact_partition("orders", PartitionId(0)).await?, 1);
        assert_eq!(
            log.read("orders", PartitionId(0), 1, 10).await?,
            vec![second.clone(), replacement]
        );
        let claim = log
            .claim_partition("orders", "billing", "member-a", PartitionId(0))
            .await?
            .ok_or_else(|| anyhow::anyhow!("member-a did not claim its partition"))?;
        assert!(log.renew_claim(&claim).await?);
        assert!(log.commit_claimed_offset(&claim, 4).await?);
        assert!(log
            .claim_partition("orders", "billing", "member-b", PartitionId(0))
            .await?
            .is_none());
        log.leave_group("orders", "billing", "member-a").await?;
        assert!(log
            .claim_partition("orders", "billing", "member-b", PartitionId(0))
            .await?
            .is_some());
        Ok(())
    }
}

#[cfg(all(test, feature = "postgres"))]
mod postgres_tests {
    use super::*;

    #[tokio::test]
    async fn append_and_replay_keep_partition_offsets() -> anyhow::Result<()> {
        let url = std::env::var("LATER_POSTGRES_TEST_URL")
            .unwrap_or_else(|_| "postgres://test:test@127.0.0.1:55432/later_test".to_owned());
        let log = PostgresLog::connect(format!("log-test-{}", crate::generate_id()), &url).await?;
        log.register_topic(TopicConfig::new("orders", 1)?).await?;
        let first = log
            .append("orders", PartitionKey::from("one"), b"first".to_vec())
            .await?;
        let second = log
            .append("orders", PartitionKey::from("two"), b"second".to_vec())
            .await?;
        assert_eq!((first.offset, second.offset), (1, 2));
        assert_eq!(
            log.read("orders", PartitionId(0), 1, 10).await?,
            vec![first.clone(), second.clone()]
        );
        assert_eq!(
            log.committed_offset("orders", "billing", PartitionId(0))
                .await?,
            1
        );
        log.commit_offset("orders", "billing", PartitionId(0), 3)
            .await?;
        log.commit_offset("orders", "billing", PartitionId(0), 2)
            .await?;
        assert_eq!(
            log.committed_offset("orders", "billing", PartitionId(0))
                .await?,
            3
        );
        let replacement = log
            .append("orders", PartitionKey::from("one"), b"replacement".to_vec())
            .await?;
        log.commit_offset("orders", "billing", PartitionId(0), 4)
            .await?;
        assert_eq!(log.compact_partition("orders", PartitionId(0)).await?, 1);
        assert_eq!(
            log.read("orders", PartitionId(0), 1, 10).await?,
            vec![second.clone(), replacement]
        );
        let claim = log
            .claim_partition("orders", "billing", "member-a", PartitionId(0))
            .await?
            .ok_or_else(|| anyhow::anyhow!("member-a did not claim its partition"))?;
        assert!(log.renew_claim(&claim).await?);
        assert!(log.commit_claimed_offset(&claim, 4).await?);
        assert!(log
            .claim_partition("orders", "billing", "member-b", PartitionId(0))
            .await?
            .is_none());
        log.leave_group("orders", "billing", "member-a").await?;
        assert!(log
            .claim_partition("orders", "billing", "member-b", PartitionId(0))
            .await?
            .is_some());
        Ok(())
    }
    #[tokio::test]
    async fn consumer_lag_reports_each_groups_own_distance_from_the_end() -> anyhow::Result<()> {
        let url = std::env::var("LATER_POSTGRES_TEST_URL")
            .unwrap_or_else(|_| "postgres://test:test@127.0.0.1:55432/later_test".to_owned());
        let log = PostgresLog::connect(format!("log-test-{}", crate::generate_id()), &url).await?;
        log.register_topic(TopicConfig::new("events", 1)?).await?;

        assert_eq!(log.consumer_lag("events").await?, Vec::new());

        for index in 1..=5 {
            log.append(
                "events",
                PartitionKey::from(format!("event-{index}")),
                format!("payload-{index}").into_bytes(),
            )
            .await?;
        }
        log.commit_offset("events", "caught-up", PartitionId(0), 6)
            .await?;
        log.commit_offset("events", "behind", PartitionId(0), 3)
            .await?;

        let mut lag = log.consumer_lag("events").await?;
        lag.sort_by(|a, b| a.group.cmp(&b.group));
        assert_eq!(
            lag,
            vec![
                ConsumerLag {
                    partition: PartitionId(0),
                    group: "behind".to_string(),
                    end_offset: 6,
                    committed_offset: 3,
                    lag: 3,
                },
                ConsumerLag {
                    partition: PartitionId(0),
                    group: "caught-up".to_string(),
                    end_offset: 6,
                    committed_offset: 6,
                    lag: 0,
                },
            ]
        );
        Ok(())
    }
}