use serde::{Deserialize, Serialize};

/// State machine for release lifecycle.
///
/// Transitions are enforced by `ReleaseState::transition()` -- the only
/// code path allowed to advance state.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum ReleaseState {
    /// Config manager has created the release and is selecting changesets.
    DraftRelease,
    /// Composition job is running (merging changesets onto the integration branch in order).
    Assembling,
    /// Composition succeeded and all tests passed.
    Validated,
    /// Git tag created, integration branch ref updated. Immutable from here on.
    Published,
    /// Deployed to at least one but not all environments.
    DeployedPartial,
    /// Deployed to every configured environment.
    DeployedFull,
    /// Release was rolled back (revert commit + new release, or prior tag redeployed).
    RolledBack,
}

impl std::fmt::Display for ReleaseState {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        let s = match self {
            Self::DraftRelease => "draft_release",
            Self::Assembling => "assembling",
            Self::Validated => "validated",
            Self::Published => "published",
            Self::DeployedPartial => "deployed_partial",
            Self::DeployedFull => "deployed_full",
            Self::RolledBack => "rolled_back",
        };
        write!(f, "{s}")
    }
}

use bson::oid::ObjectId;
use chrono::{DateTime, Utc};

/// A release batch: a curated, ordered subset of queued changesets
/// that will be composed into a single tagged commit on the integration branch.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ReleaseBatch {
    /// MongoDB document ID.
    #[serde(rename = "_id")]
    pub id: ObjectId,

    /// The app this release belongs to.
    pub app_id: ObjectId,

    /// Release tag in `rYYYY.MM.DD.N` format. Assigned at creation time
    /// (next available sequence number for today). Unique per app.
    pub tag: String,

    /// Current lifecycle state.
    pub state: ReleaseState,

    /// Changeset IDs in composition order. Position is implicit (vec index).
    /// Only changesets in `queued` state may be added.
    pub ordered_changeset_ids: Vec<ObjectId>,

    /// Job ID of the composition/assembly job (set when assembly starts).
    pub compose_job_id: Option<ObjectId>,

    /// SHA of the final composed commit on the integration branch (set after publish).
    pub published_sha: Option<String>,

    /// Timestamp when the release was published.
    pub published_at: Option<DateTime<Utc>>,

    /// User ID of the actor who triggered publish.
    pub published_by: Option<ObjectId>,

    /// When the draft was first created.
    pub created_at: DateTime<Utc>,

    /// Last modification timestamp.
    pub updated_at: DateTime<Utc>,
}

/// Tracks per-changeset state within a release composition.
///
/// One document per changeset included in a release. The `position` field
/// determines merge order during assembly.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ReleaseChangeset {
    /// MongoDB document ID.
    #[serde(rename = "_id")]
    pub id: ObjectId,

    /// Parent release batch.
    pub release_id: ObjectId,

    /// The changeset being included.
    pub changeset_id: ObjectId,

    /// Zero-based position in the merge order.
    pub position: u32,

    /// SHA of the merge commit created for this specific changeset during
    /// composition. Set by the assembly worker after a successful merge step.
    pub merge_sha: Option<String>,
}

use crate::error::ConmanError;

impl ReleaseState {
    /// Attempt to transition from the current state to `target`.
    ///
    /// Guard conditions are checked inline. Returns the new state on
    /// success or `ConmanError::InvalidTransition` on failure.
    pub fn transition(
        self,
        target: ReleaseState,
        guard: &TransitionGuard,
    ) -> Result<ReleaseState, ConmanError> {
        use ReleaseState::*;

        let allowed = match (self, target) {
            // Draft -> Assembling: must have at least one changeset selected.
            (DraftRelease, Assembling) => guard.has_changesets,

            // Assembling -> Validated: compose job succeeded with no conflicts
            // or test failures.
            (Assembling, Validated) => guard.compose_succeeded,

            // Assembling -> DraftRelease: compose failed (conflict or test
            // failure), config manager can revise the selection.
            (Assembling, DraftRelease) => guard.compose_failed,

            // Validated -> Published: tag created and integration branch ref updated.
            (Validated, Published) => guard.tag_created,

            // Published -> DeployedPartial: first deployment to any env succeeded.
            (Published, DeployedPartial) => true,

            // DeployedPartial -> DeployedFull: all environments deployed.
            (DeployedPartial, DeployedFull) => guard.all_envs_deployed,

            // DeployedPartial -> DeployedPartial: another env deployed but not all.
            (DeployedPartial, DeployedPartial) => !guard.all_envs_deployed,

            // Published or DeployedPartial or DeployedFull -> RolledBack.
            (Published, RolledBack)
            | (DeployedPartial, RolledBack)
            | (DeployedFull, RolledBack) => true,

            _ => false,
        };

        if allowed {
            Ok(target)
        } else {
            Err(ConmanError::InvalidTransition {
                from: self.to_string(),
                to: target.to_string(),
            })
        }
    }
}

/// Guard conditions evaluated before a state transition is accepted.
///
/// Populated by the caller (handler or job worker) from current domain state.
#[derive(Debug, Clone, Default)]
pub struct TransitionGuard {
    /// At least one changeset is in `ordered_changeset_ids`.
    pub has_changesets: bool,
    /// Compose job completed without conflicts or test failures.
    pub compose_succeeded: bool,
    /// Compose job failed (conflicts or tests).
    pub compose_failed: bool,
    /// Git tag was created and integration branch ref updated.
    pub tag_created: bool,
    /// All configured environments have a successful deployment for this release.
    pub all_envs_deployed: bool,
}

use bson::oid::ObjectId;
use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};

/// POST /api/apps/:appId/releases
///
/// Creates a draft release. The tag is auto-generated as `rYYYY.MM.DD.N`.
/// Optionally accepts an initial set of changeset IDs to include.
#[derive(Debug, Deserialize)]
pub struct CreateReleaseRequest {
    /// Optional initial changeset IDs to include (must all be in `queued` state).
    #[serde(default)]
    pub changeset_ids: Vec<String>,
}

/// POST /api/apps/:appId/releases/:releaseId/changesets
///
/// Add or remove changesets from a draft release. Only valid when release
/// is in `draft_release` state.
#[derive(Debug, Deserialize)]
pub struct AddChangesetsRequest {
    /// Changeset IDs to add (must be in `queued` state).
    #[serde(default)]
    pub add: Vec<String>,
    /// Changeset IDs to remove from the release.
    #[serde(default)]
    pub remove: Vec<String>,
}

/// POST /api/apps/:appId/releases/:releaseId/reorder
///
/// Set the explicit merge order for changesets in a draft release.
#[derive(Debug, Deserialize)]
pub struct ReorderRequest {
    /// Changeset IDs in the desired merge order. Must be a permutation of
    /// the current `ordered_changeset_ids`.
    pub ordered_changeset_ids: Vec<String>,
}

/// Response DTO returned for release detail and list endpoints.
#[derive(Debug, Serialize)]
pub struct ReleaseResponse {
    pub id: String,
    pub app_id: String,
    pub tag: String,
    pub state: String,
    pub ordered_changeset_ids: Vec<String>,
    pub compose_job_id: Option<String>,
    pub published_sha: Option<String>,
    pub published_at: Option<DateTime<Utc>>,
    pub published_by: Option<String>,
    pub created_at: DateTime<Utc>,
    pub updated_at: DateTime<Utc>,
}

/// List query parameters for GET /api/apps/:appId/releases
#[derive(Debug, Deserialize)]
pub struct ReleaseListQuery {
    #[serde(default = "super::default_page")]
    pub page: u64,
    #[serde(default = "super::default_limit")]
    pub limit: u64,
    /// Optional state filter (e.g. `?state=draft_release`).
    pub state: Option<String>,
}

impl From<&ReleaseBatch> for ReleaseResponse {
    fn from(r: &ReleaseBatch) -> Self {
        Self {
            id: r.id.to_hex(),
            app_id: r.app_id.to_hex(),
            tag: r.tag.clone(),
            state: r.state.to_string(),
            ordered_changeset_ids: r.ordered_changeset_ids.iter().map(|id| id.to_hex()).collect(),
            compose_job_id: r.compose_job_id.map(|id| id.to_hex()),
            published_sha: r.published_sha.clone(),
            published_at: r.published_at,
            published_by: r.published_by.map(|id| id.to_hex()),
            created_at: r.created_at,
            updated_at: r.updated_at,
        }
    }
}

/// Indexes for the release_batches collection.
async fn ensure_indexes(&self) -> Result<(), ConmanError> {
    let collection = self.db.collection::<ReleaseBatch>("release_batches");

    // Lookup by app + state (list releases filtered by state).
    collection.create_index(
        IndexModel::builder()
            .keys(doc! { "app_id": 1, "state": 1 })
            .build(),
    ).await?;

    // Unique tag per app -- prevents duplicate release tags.
    collection.create_index(
        IndexModel::builder()
            .keys(doc! { "app_id": 1, "tag": 1 })
            .options(IndexOptions::builder().unique(true).build())
            .build(),
    ).await?;

    // Lookup by app sorted by creation time (list recent releases).
    collection.create_index(
        IndexModel::builder()
            .keys(doc! { "app_id": 1, "created_at": -1 })
            .build(),
    ).await?;

    Ok(())
}

{
  "_id": ObjectId("664f1a2b3c4d5e6f7a8b9c0e"),
  "app_id": ObjectId("664f1a2b3c4d5e6f7a8b9c01"),
  "tag": "r2026.02.25.1",
  "state": "draft_release",
  "ordered_changeset_ids": [
    ObjectId("664f1a2b3c4d5e6f7a8b9c10"),
    ObjectId("664f1a2b3c4d5e6f7a8b9c11")
  ],
  "compose_job_id": null,
  "published_sha": null,
  "published_at": null,
  "published_by": null,
  "created_at": ISODate("2026-02-25T10:00:00Z"),
  "updated_at": ISODate("2026-02-25T10:05:00Z")
}

{
  "_id": ObjectId("664f1a2b3c4d5e6f7a8b9c0f"),
  "app_id": ObjectId("664f1a2b3c4d5e6f7a8b9c01"),
  "tag": "r2026.02.24.2",
  "state": "published",
  "ordered_changeset_ids": [
    ObjectId("664f1a2b3c4d5e6f7a8b9c12"),
    ObjectId("664f1a2b3c4d5e6f7a8b9c13"),
    ObjectId("664f1a2b3c4d5e6f7a8b9c14")
  ],
  "compose_job_id": ObjectId("664f1a2b3c4d5e6f7a8b9c20"),
  "published_sha": "a1b2c3d4e5f6a1b2c3d4e5f6a1b2c3d4e5f6a1b2",
  "published_at": ISODate("2026-02-24T16:42:00Z"),
  "published_by": ObjectId("664f1a2b3c4d5e6f7a8b9c02"),
  "created_at": ISODate("2026-02-24T14:00:00Z"),
  "updated_at": ISODate("2026-02-24T16:42:00Z")
}

/// Indexes for the release_changesets collection.
async fn ensure_indexes(&self) -> Result<(), ConmanError> {
    let collection = self.db.collection::<ReleaseChangeset>("release_changesets");

    // Ordered lookup of all changesets in a release.
    collection.create_index(
        IndexModel::builder()
            .keys(doc! { "release_id": 1, "position": 1 })
            .build(),
    ).await?;

    // Reverse lookup: which release(s) include a given changeset.
    // A changeset may only appear in one non-draft release, but this
    // index supports the check.
    collection.create_index(
        IndexModel::builder()
            .keys(doc! { "changeset_id": 1 })
            .build(),
    ).await?;

    Ok(())
}

{
  "_id": ObjectId("664f1a2b3c4d5e6f7a8b9c30"),
  "release_id": ObjectId("664f1a2b3c4d5e6f7a8b9c0f"),
  "changeset_id": ObjectId("664f1a2b3c4d5e6f7a8b9c12"),
  "position": 0,
  "merge_sha": "b2c3d4e5f6a1b2c3d4e5f6a1b2c3d4e5f6a1b2c3"
}

{
  "data": [
    {
      "id": "664f1a2b3c4d5e6f7a8b9c0e",
      "app_id": "664f1a2b3c4d5e6f7a8b9c01",
      "tag": "r2026.02.25.1",
      "state": "draft_release",
      "ordered_changeset_ids": ["664f1a2b3c4d5e6f7a8b9c10"],
      "compose_job_id": null,
      "published_sha": null,
      "published_at": null,
      "published_by": null,
      "created_at": "2026-02-25T10:00:00Z",
      "updated_at": "2026-02-25T10:05:00Z"
    }
  ],
  "pagination": { "page": 1, "limit": 20, "total": 1 }
}

/// GET /api/apps/:appId/releases
///
/// List releases for the app, optionally filtered by state.
/// Sorted by created_at descending (most recent first).
pub async fn list_releases(
    State(state): State<AppState>,
    Extension(auth_user): Extension<AuthUser>,
    Path(app_id): Path<String>,
    Query(params): Query<ReleaseListQuery>,
) -> Result<Json<ApiResponse<Vec<ReleaseResponse>>>, ConmanError> {
    let app_id = parse_object_id(&app_id)?;
    auth_user.require_member(app_id)?;

    let pagination = Pagination { page: params.page, limit: params.limit }.validate()?;
    let state_filter = params.state.as_deref().map(parse_release_state).transpose()?;

    let (releases, total) = release_repo
        .list_by_app(app_id, state_filter, &pagination)
        .await?;

    let data: Vec<ReleaseResponse> = releases.iter().map(ReleaseResponse::from).collect();
    Ok(Json(ApiResponse::paginated(data, pagination.page, pagination.limit, total)))
}

{
  "changeset_ids": ["664f1a2b3c4d5e6f7a8b9c10", "664f1a2b3c4d5e6f7a8b9c11"]
}

{
  "data": {
    "id": "664f1a2b3c4d5e6f7a8b9c0e",
    "app_id": "664f1a2b3c4d5e6f7a8b9c01",
    "tag": "r2026.02.25.1",
    "state": "draft_release",
    "ordered_changeset_ids": ["664f1a2b3c4d5e6f7a8b9c10", "664f1a2b3c4d5e6f7a8b9c11"],
    "compose_job_id": null,
    "published_sha": null,
    "published_at": null,
    "published_by": null,
    "created_at": "2026-02-25T10:00:00Z",
    "updated_at": "2026-02-25T10:00:00Z"
  }
}

{
  "data": {
    "id": "664f1a2b3c4d5e6f7a8b9c0e",
    "app_id": "664f1a2b3c4d5e6f7a8b9c01",
    "tag": "r2026.02.25.1",
    "state": "draft_release",
    "ordered_changeset_ids": ["664f1a2b3c4d5e6f7a8b9c10"],
    "changesets": [
      {
        "changeset_id": "664f1a2b3c4d5e6f7a8b9c10",
        "position": 0,
        "merge_sha": null
      }
    ],
    "compose_job_id": null,
    "published_sha": null,
    "published_at": null,
    "published_by": null,
    "created_at": "2026-02-25T10:00:00Z",
    "updated_at": "2026-02-25T10:05:00Z"
  }
}

{
  "add": ["664f1a2b3c4d5e6f7a8b9c15"],
  "remove": ["664f1a2b3c4d5e6f7a8b9c11"]
}

{
  "ordered_changeset_ids": [
    "664f1a2b3c4d5e6f7a8b9c11",
    "664f1a2b3c4d5e6f7a8b9c10"
  ]
}

{
  "data": {
    "id": "664f1a2b3c4d5e6f7a8b9c0e",
    "state": "assembling",
    "compose_job_id": "664f1a2b3c4d5e6f7a8b9c20",
    "tag": "r2026.02.25.1"
  }
}

/// Compose a release by merging each selected changeset onto the integration branch in order.
///
/// For each changeset (in position order):
///   1. Merge the changeset branch into a temp ref using UserMergeToRef.
///   2. If merge conflicts → mark changeset as `conflicted`, fail the job.
///   3. If msuite test fails → mark changeset as `needs_revalidation`, fail the job.
///   4. Record merge_sha on the ReleaseChangeset document.
///
/// On full success:
///   - Transition release to Validated.
///
/// On failure:
///   - Transition release back to DraftRelease.
///   - Mark failing changeset(s) with appropriate state.
async fn execute_release_assemble(job: &Job, ctx: &WorkerContext) -> Result<(), ConmanError> {
    let release = ctx.release_repo.find_by_id(job.entity_id).await?;
    let app = ctx.app_repo.find_by_id(release.app_id).await?;
    let repo = app_to_gitaly_repo(&app);

    // Resolve current integration branch HEAD as the starting point.
    let integration_commit = ctx
        .gitaly
        .find_commit(&repo, "refs/heads/<integration_branch>")
        .await?;
    let mut current_sha = integration_commit.id.clone();

    // Create a temporary composition ref to avoid touching integration branch until publish.
    let compose_ref = format!("refs/conman/compose/{}", release.id.to_hex());

    let release_changesets = ctx.release_changeset_repo
        .find_by_release_ordered(release.id)
        .await?;

    for rc in &release_changesets {
        let changeset = ctx.changeset_repo.find_by_id(rc.changeset_id).await?;

        // Merge changeset head into the running composition ref.
        let merge_result = ctx.gitaly.user_merge_to_ref(
            &repo,
            &ctx.conman_user(),
            &changeset.head_sha,
            compose_ref.as_bytes(),   // target_ref
            current_sha.as_bytes(),    // first_parent_ref resolved from prior step
            format!(
                "Compose changeset {} into release {}",
                changeset.id.to_hex(),
                release.tag
            ).as_bytes(),
        ).await;

        match merge_result {
            Ok(response) => {
                // Record the merge SHA for this changeset.
                ctx.release_changeset_repo
                    .set_merge_sha(rc.id, &response.commit_id)
                    .await?;
                current_sha = response.commit_id;
            }
            Err(e) if e.is_merge_conflict() => {
                // Mark this changeset as conflicted.
                ctx.changeset_repo
                    .transition_state(rc.changeset_id, ChangesetState::Conflicted)
                    .await?;

                // Fail the release back to draft.
                ctx.release_repo
                    .transition_state(release.id, ReleaseState::DraftRelease)
                    .await?;

                return Err(ConmanError::Git {
                    message: format!(
                        "Merge conflict composing changeset {}",
                        changeset.id.to_hex()
                    ),
                });
            }
            Err(e) => return Err(e),
        }
    }

    // All merges succeeded -- run msuite validation on the composed result.
    // (Delegates to msuite_merge job or inline check depending on config.)

    // Transition release to Validated.
    ctx.release_repo
        .transition_state(release.id, ReleaseState::Validated)
        .await?;

    Ok(())
}

{
  "data": {
    "id": "664f1a2b3c4d5e6f7a8b9c0e",
    "state": "published",
    "tag": "r2026.02.25.1",
    "published_sha": "a1b2c3d4e5f6a1b2c3d4e5f6a1b2c3d4e5f6a1b2",
    "published_at": "2026-02-25T16:42:00Z",
    "published_by": "664f1a2b3c4d5e6f7a8b9c02"
  }
}

/// Generate the next release tag for today.
///
/// Queries existing tags matching today's date prefix and increments the
/// sequence number. Thread-safe because the unique index on (app_id, tag)
/// rejects duplicates, causing a retry with the next N.
pub async fn next_tag(
    release_repo: &ReleaseRepo,
    app_id: ObjectId,
) -> Result<String, ConmanError> {
    let today = Utc::now().format("%Y.%m.%d").to_string();
    let prefix = format!("r{today}.");

    // Find the highest N for today's tags on this app.
    let max_n = release_repo
        .find_max_tag_sequence(app_id, &prefix)
        .await?;

    let next_n = max_n.map_or(1, |n| n + 1);
    Ok(format!("r{today}.{next_n}"))
}

// Two-phase streaming merge. First request sends parameters, first response
// returns the merge commit ID. Second request sets apply=true to commit the
// ref update. Executes hooks and authorization checks.
rpc UserMergeBranch(stream UserMergeBranchRequest) returns (stream UserMergeBranchResponse);

message UserMergeBranchRequest {
  // Repository where the merge happens.
  Repository repository = 1;
  // User performing the operation (auth + commit author).
  User user = 2;
  // Object ID of the commit to merge into the target branch.
  string commit_id = 3;
  // Target branch name (e.g. app integration branch).
  bytes branch = 4;
  // Merge commit message.
  bytes message = 5;
  // Set to true in the second message to apply the merge.
  bool apply = 6;
  // Optional timestamp for the merge commit.
  google.protobuf.Timestamp timestamp = 7;
  // Expected current OID of the branch (optimistic lock).
  string expected_old_oid = 8;
  // If true, fast-forward the squash commit instead of creating a merge commit.
  bool squash = 9;
  // Whether to sign the commit.
  bool sign = 10;
}

message UserMergeBranchResponse {
  // Merge commit OID (returned in the first response).
  string commit_id = 1;
  // Branch update details (returned in the second response after apply).
  OperationBranchUpdate branch_update = 3;
}

message OperationBranchUpdate {
  // OID of the commit the branch now points to.
  string commit_id = 1;
  // Whether this was the first branch in the repo.
  bool repo_created = 2;
  // Whether the branch was newly created (vs. updated).
  bool branch_created = 3;
}

message UserMergeBranchError {
  oneof error {
    AccessCheckError access_check = 1;
    ReferenceUpdateError reference_update = 2;
    CustomHookError custom_hook = 3;
    MergeConflictError merge_conflict = 4;
  }
}

/// Fast-forward the integration branch to the composed commit using the
/// two-phase merge RPC.
///
/// Sets expected_old_oid to the current integration branch HEAD to prevent races.
/// If another release or process updated integration branch between validation and
/// publish, this will fail with a ReferenceUpdateError, which the
/// handler surfaces as a ConmanError::Conflict.
pub async fn merge_to_integration(
    &self,
    repo: &Repository,
    user: &User,
    integration_branch: &str,
    commit_id: &str,
    expected_integration_oid: &str,
    message: &str,
) -> Result<String, ConmanError> {
    // Phase 1: send merge parameters.
    let req1 = UserMergeBranchRequest {
        repository: Some(repo.clone()),
        user: Some(user.clone()),
        commit_id: commit_id.to_string(),
        branch: integration_branch.as_bytes().to_vec(),
        message: message.as_bytes().to_vec(),
        apply: false,
        expected_old_oid: expected_integration_oid.to_string(),
        ..Default::default()
    };

    // Phase 2: apply the merge.
    let req2 = UserMergeBranchRequest {
        apply: true,
        ..Default::default()
    };

    let responses = self.operation_client
        .user_merge_branch(tokio_stream::iter(vec![req1, req2]))
        .await
        .map_err(|e| self.map_grpc_error("UserMergeBranch", e))?
        .into_inner();

    // Collect both responses from the stream.
    let mut commit_id_out = String::new();
    let mut responses = responses;
    while let Some(resp) = responses.message().await.map_err(|e| {
        self.map_grpc_error("UserMergeBranch stream", e)
    })? {
        if !resp.commit_id.is_empty() {
            commit_id_out = resp.commit_id;
        }
    }

    Ok(commit_id_out)
}

// Merge source_sha into first_parent_ref and write result to target_ref.
// Does not execute hooks. Overwrites target_ref if it exists.
rpc UserMergeToRef(UserMergeToRefRequest) returns (UserMergeToRefResponse);

message UserMergeToRefRequest {
  // Repository to perform the merge in.
  Repository repository = 1;
  // User for commit authorship.
  User user = 2;
  // Object ID of the second parent (changeset head SHA).
  string source_sha = 3;
  // Deprecated; use first_parent_ref instead.
  bytes branch = 4 [deprecated = true];
  // Fully-qualified ref to write the merge commit to.
  bytes target_ref = 5;
  // Merge commit message.
  bytes message = 6;
  // Fully-qualified ref or OID used as the first parent (integration branch line).
  bytes first_parent_ref = 7;
  // Deprecated, no longer used.
  bool allow_conflicts = 8 [deprecated = true];
  // Optional timestamp for the merge commit.
  google.protobuf.Timestamp timestamp = 9;
  // Expected OID of target_ref for optimistic locking.
  string expected_old_oid = 10;
  // Whether to sign the commit.
  bool sign = 11;
}

message UserMergeToRefResponse {
  // Object ID of the created merge commit.
  string commit_id = 1;
}

/// Merge a changeset into the composition ref during release assembly.
///
/// first_parent_sha is the OID from the previous composition step
/// (or the current integration branch HEAD for the first changeset).
pub async fn merge_changeset_to_compose_ref(
    &self,
    repo: &Repository,
    user: &User,
    changeset_head_sha: &str,
    first_parent_sha: &str,
    compose_ref: &str,
    message: &str,
) -> Result<UserMergeToRefResponse, ConmanError> {
    let request = UserMergeToRefRequest {
        repository: Some(repo.clone()),
        user: Some(user.clone()),
        source_sha: changeset_head_sha.to_string(),
        target_ref: compose_ref.as_bytes().to_vec(),
        message: message.as_bytes().to_vec(),
        first_parent_ref: first_parent_sha.as_bytes().to_vec(),
        ..Default::default()
    };

    self.operation_client
        .user_merge_to_ref(request)
        .await
        .map(|r| r.into_inner())
        .map_err(|e| self.map_grpc_error("UserMergeToRef", e))
}

// Create a lightweight or annotated tag. Lightweight if message is empty.
rpc UserCreateTag(UserCreateTagRequest) returns (UserCreateTagResponse);

message UserCreateTagRequest {
  // Repository to create the tag in.
  Repository repository = 1;
  // Tag name (without refs/tags/ prefix), e.g. "r2026.02.25.1".
  bytes tag_name = 2;
  // User performing the operation.
  User user = 3;
  // Revision the tag should point to (the composed commit SHA).
  bytes target_revision = 4;
  // Tag message. Empty for lightweight tags.
  bytes message = 5;
  // Optional timestamp (only for annotated tags).
  google.protobuf.Timestamp timestamp = 7;
}

message UserCreateTagResponse {
  // The created tag object.
  Tag tag = 1;
}

message UserCreateTagError {
  oneof error {
    AccessCheckError access_check = 1;
    ReferenceUpdateError reference_update = 2;
    CustomHookError custom_hook = 3;
    ReferenceExistsError reference_exists = 4;
  }
}

/// Create a lightweight release tag pointing at the composed commit.
pub async fn create_release_tag(
    &self,
    repo: &Repository,
    user: &User,
    tag_name: &str,
    target_sha: &str,
) -> Result<Tag, ConmanError> {
    let request = UserCreateTagRequest {
        repository: Some(repo.clone()),
        user: Some(user.clone()),
        tag_name: tag_name.as_bytes().to_vec(),
        target_revision: target_sha.as_bytes().to_vec(),
        message: Vec::new(), // lightweight tag
        ..Default::default()
    };

    let response = self.operation_client
        .user_create_tag(request)
        .await
        .map_err(|e| self.map_grpc_error("UserCreateTag", e))?
        .into_inner();

    response.tag.ok_or_else(|| ConmanError::Git {
        message: "UserCreateTag returned no tag".to_string(),
    })
}

// Look up a single tag by name. Returns Internal error if not found.
rpc FindTag(FindTagRequest) returns (FindTagResponse);

message FindTagRequest {
  // Repository to look up the tag in.
  Repository repository = 1;
  // Tag name without refs/tags/ prefix (e.g. "r2026.02.25.1").
  bytes tag_name = 2;
}

message FindTagResponse {
  // The found tag object.
  Tag tag = 1;
}

message FindTagError {
  oneof error {
    // Set when the tag was not found.
    ReferenceNotFoundError tag_not_found = 1;
  }
}

/// Check whether a tag already exists in Git.
///
/// Returns Ok(Some(tag)) if found, Ok(None) if not found,
/// Err for unexpected gRPC failures.
pub async fn find_tag(
    &self,
    repo: &Repository,
    tag_name: &str,
) -> Result<Option<Tag>, ConmanError> {
    let request = FindTagRequest {
        repository: Some(repo.clone()),
        tag_name: tag_name.as_bytes().to_vec(),
    };

    match self.ref_client.find_tag(request).await {
        Ok(response) => Ok(response.into_inner().tag),
        Err(status) if status.code() == tonic::Code::Internal => {
            // FindTag returns Internal when tag is not found.
            Ok(None)
        }
        Err(e) => Err(self.map_grpc_error("FindTag", e)),
    }
}

// Stream all tags under refs/tags/ for a repository.
rpc FindAllTags(FindAllTagsRequest) returns (stream FindAllTagsResponse);

message FindAllTagsRequest {
  message SortBy {
    enum Key {
      REFNAME = 0;
      CREATORDATE = 1;
      VERSION_REFNAME = 2;
    }
    Key key = 1;
    SortDirection direction = 2;
  }

  // Repository to list tags from.
  Repository repository = 1;
  // Optional sort order.
  SortBy sort_by = 2;
  // Optional pagination.
  PaginationParameter pagination_params = 3;
}

message FindAllTagsResponse {
  // List of tags in this chunk.
  repeated Tag tags = 1;
}

/// List all tags in the repository, collecting them from the response stream.
///
/// Used to find existing release tags for sequence number generation.
/// Filters client-side by the `rYYYY.MM.DD.` prefix for today's date.
pub async fn list_all_tags(
    &self,
    repo: &Repository,
) -> Result<Vec<Tag>, ConmanError> {
    let request = FindAllTagsRequest {
        repository: Some(repo.clone()),
        sort_by: None,
        pagination_params: None,
    };

    let mut stream = self.ref_client
        .find_all_tags(request)
        .await
        .map_err(|e| self.map_grpc_error("FindAllTags", e))?
        .into_inner();

    let mut tags = Vec::new();
    while let Some(response) = stream.message().await.map_err(|e| {
        self.map_grpc_error("FindAllTags stream", e)
    })? {
        tags.extend(response.tags);
    }

    Ok(tags)
}

// Find a commit by commitish. Returns nil commit if not found.
rpc FindCommit(FindCommitRequest) returns (FindCommitResponse);

message FindCommitRequest {
  // Repository to search in.
  Repository repository = 1;
  // Commitish to resolve (e.g. "refs/heads/<integration_branch>", a SHA, a tag name).
  bytes revision = 2;
  // If true, parse and include Git trailers.
  bool trailers = 3;
}

message FindCommitResponse {
  // The resolved commit, or nil if not found.
  GitCommit commit = 1;
}

/// Resolve a commitish to a full GitCommit object.
///
/// Returns NotFound if the revision does not resolve to a commit.
pub async fn find_commit(
    &self,
    repo: &Repository,
    revision: &str,
) -> Result<GitCommit, ConmanError> {
    let request = FindCommitRequest {
        repository: Some(repo.clone()),
        revision: revision.as_bytes().to_vec(),
        trailers: false,
    };

    let response = self.commit_client
        .find_commit(request)
        .await
        .map_err(|e| self.map_grpc_error("FindCommit", e))?
        .into_inner();

    response.commit.ok_or_else(|| ConmanError::NotFound {
        entity: "commit",
        id: revision.to_string(),
    })
}

#[test]
fn draft_to_assembling_with_changesets() {
    let guard = TransitionGuard { has_changesets: true, ..Default::default() };
    let result = ReleaseState::DraftRelease.transition(ReleaseState::Assembling, &guard);
    assert_eq!(result.unwrap(), ReleaseState::Assembling);
}

#[test]
fn assembling_to_validated_on_success() {
    let guard = TransitionGuard { compose_succeeded: true, ..Default::default() };
    let result = ReleaseState::Assembling.transition(ReleaseState::Validated, &guard);
    assert_eq!(result.unwrap(), ReleaseState::Validated);
}

#[test]
fn assembling_to_draft_on_failure() {
    let guard = TransitionGuard { compose_failed: true, ..Default::default() };
    let result = ReleaseState::Assembling.transition(ReleaseState::DraftRelease, &guard);
    assert_eq!(result.unwrap(), ReleaseState::DraftRelease);
}

#[test]
fn validated_to_published() {
    let guard = TransitionGuard { tag_created: true, ..Default::default() };
    let result = ReleaseState::Validated.transition(ReleaseState::Published, &guard);
    assert_eq!(result.unwrap(), ReleaseState::Published);
}

#[test]
fn published_to_deployed_partial() {
    let guard = TransitionGuard::default();
    let result = ReleaseState::Published.transition(ReleaseState::DeployedPartial, &guard);
    assert_eq!(result.unwrap(), ReleaseState::DeployedPartial);
}

#[test]
fn deployed_partial_to_deployed_full() {
    let guard = TransitionGuard { all_envs_deployed: true, ..Default::default() };
    let result = ReleaseState::DeployedPartial.transition(ReleaseState::DeployedFull, &guard);
    assert_eq!(result.unwrap(), ReleaseState::DeployedFull);
}

#[test]
fn deployed_partial_stays_partial_when_not_all_envs() {
    let guard = TransitionGuard { all_envs_deployed: false, ..Default::default() };
    let result = ReleaseState::DeployedPartial.transition(ReleaseState::DeployedPartial, &guard);
    assert_eq!(result.unwrap(), ReleaseState::DeployedPartial);
}

#[test]
fn published_to_rolled_back() {
    let guard = TransitionGuard::default();
    let result = ReleaseState::Published.transition(ReleaseState::RolledBack, &guard);
    assert_eq!(result.unwrap(), ReleaseState::RolledBack);
}

#[test]
fn deployed_full_to_rolled_back() {
    let guard = TransitionGuard::default();
    let result = ReleaseState::DeployedFull.transition(ReleaseState::RolledBack, &guard);
    assert_eq!(result.unwrap(), ReleaseState::RolledBack);
}

#[test]
fn draft_to_assembling_without_changesets_is_rejected() {
    let guard = TransitionGuard { has_changesets: false, ..Default::default() };
    let result = ReleaseState::DraftRelease.transition(ReleaseState::Assembling, &guard);
    assert!(result.is_err());
    assert!(result.unwrap_err().to_string().contains("invalid state transition"));
}

#[test]
fn draft_to_published_is_rejected() {
    let guard = TransitionGuard::default();
    let result = ReleaseState::DraftRelease.transition(ReleaseState::Published, &guard);
    assert!(result.is_err());
}

#[test]
fn published_to_draft_is_rejected() {
    let guard = TransitionGuard::default();
    let result = ReleaseState::Published.transition(ReleaseState::DraftRelease, &guard);
    assert!(result.is_err());
}

#[test]
fn published_to_assembling_is_rejected() {
    let guard = TransitionGuard::default();
    let result = ReleaseState::Published.transition(ReleaseState::Assembling, &guard);
    assert!(result.is_err());
}

#[test]
fn validated_to_draft_is_rejected() {
    let guard = TransitionGuard::default();
    let result = ReleaseState::Validated.transition(ReleaseState::DraftRelease, &guard);
    assert!(result.is_err());
}

#[test]
fn rolled_back_to_any_is_rejected() {
    let guard = TransitionGuard { tag_created: true, has_changesets: true, ..Default::default() };
    for target in [
        ReleaseState::DraftRelease,
        ReleaseState::Assembling,
        ReleaseState::Validated,
        ReleaseState::Published,
        ReleaseState::DeployedPartial,
        ReleaseState::DeployedFull,
    ] {
        let result = ReleaseState::RolledBack.transition(target, &guard);
        assert!(result.is_err(), "RolledBack -> {target} should be rejected");
    }
}

#[tokio::test]
async fn first_tag_of_the_day_is_1() {
    let repo = test_release_repo().await;
    let tag = next_tag(&repo, test_app_id()).await.unwrap();
    let today = Utc::now().format("%Y.%m.%d");
    assert_eq!(tag, format!("r{today}.1"));
}

#[tokio::test]
async fn second_tag_of_the_day_is_2() {
    let repo = test_release_repo().await;
    let app_id = test_app_id();

    // Insert a release with today's first tag.
    insert_release(&repo, app_id, &format!("r{}.1", Utc::now().format("%Y.%m.%d"))).await;

    let tag = next_tag(&repo, app_id).await.unwrap();
    let today = Utc::now().format("%Y.%m.%d");
    assert_eq!(tag, format!("r{today}.2"));
}

#[tokio::test]
async fn tags_from_different_days_dont_affect_sequence() {
    let repo = test_release_repo().await;
    let app_id = test_app_id();

    // Insert a release from yesterday.
    insert_release(&repo, app_id, "r2026.02.24.5").await;

    let tag = next_tag(&repo, app_id).await.unwrap();
    let today = Utc::now().format("%Y.%m.%d");
    assert_eq!(tag, format!("r{today}.1"));
}

#[tokio::test]
async fn create_release_returns_201_with_tag() {
    let app = test_app_with_real_mongo().await;
    let app_id = seed_app(&app).await;
    let cs1 = seed_queued_changeset(&app, app_id).await;
    let cs2 = seed_queued_changeset(&app, app_id).await;

    let response = app.oneshot(
        Request::builder()
            .method("POST")
            .uri(&format!("/api/apps/{app_id}/releases"))
            .header("content-type", "application/json")
            .header("authorization", config_manager_token())
            .body(Body::from(json!({
                "changeset_ids": [cs1.to_hex(), cs2.to_hex()]
            }).to_string()))
            .unwrap()
    ).await.unwrap();

    assert_eq!(response.status(), StatusCode::CREATED);

    let body: serde_json::Value = parse_body(response).await;
    assert_eq!(body["data"]["state"], "draft_release");
    assert!(body["data"]["tag"].as_str().unwrap().starts_with("r"));
    assert_eq!(body["data"]["ordered_changeset_ids"].as_array().unwrap().len(), 2);
}

#[tokio::test]
async fn create_release_rejects_non_queued_changesets() {
    let app = test_app_with_real_mongo().await;
    let app_id = seed_app(&app).await;
    let draft_cs = seed_draft_changeset(&app, app_id).await;

    let response = app.oneshot(
        Request::builder()
            .method("POST")
            .uri(&format!("/api/apps/{app_id}/releases"))
            .header("content-type", "application/json")
            .header("authorization", config_manager_token())
            .body(Body::from(json!({
                "changeset_ids": [draft_cs.to_hex()]
            }).to_string()))
            .unwrap()
    ).await.unwrap();

    assert_eq!(response.status(), StatusCode::BAD_REQUEST);
}

#[tokio::test]
async fn create_release_rejects_user_role() {
    let app = test_app_with_real_mongo().await;
    let app_id = seed_app(&app).await;

    let response = app.oneshot(
        Request::builder()
            .method("POST")
            .uri(&format!("/api/apps/{app_id}/releases"))
            .header("content-type", "application/json")
            .header("authorization", user_token()) // not config_manager
            .body(Body::from(json!({ "changeset_ids": [] }).to_string()))
            .unwrap()
    ).await.unwrap();

    assert_eq!(response.status(), StatusCode::FORBIDDEN);
}

#[tokio::test]
async fn add_changeset_to_draft_release() {
    let app = test_app_with_real_mongo().await;
    let (app_id, release_id) = seed_draft_release(&app).await;
    let cs = seed_queued_changeset(&app, app_id).await;

    let response = app.oneshot(
        Request::builder()
            .method("POST")
            .uri(&format!("/api/apps/{app_id}/releases/{release_id}/changesets"))
            .header("content-type", "application/json")
            .header("authorization", config_manager_token())
            .body(Body::from(json!({
                "add": [cs.to_hex()],
                "remove": []
            }).to_string()))
            .unwrap()
    ).await.unwrap();

    assert_eq!(response.status(), StatusCode::OK);
    let body: serde_json::Value = parse_body(response).await;
    assert!(body["data"]["ordered_changeset_ids"]
        .as_array().unwrap()
        .iter()
        .any(|id| id.as_str() == Some(&cs.to_hex())));
}

#[tokio::test]
async fn modify_changesets_rejected_for_non_draft_release() {
    let app = test_app_with_real_mongo().await;
    let (app_id, release_id) = seed_published_release(&app).await;
    let cs = seed_queued_changeset(&app, app_id).await;

    let response = app.oneshot(
        Request::builder()
            .method("POST")
            .uri(&format!("/api/apps/{app_id}/releases/{release_id}/changesets"))
            .header("content-type", "application/json")
            .header("authorization", config_manager_token())
            .body(Body::from(json!({
                "add": [cs.to_hex()],
                "remove": []
            }).to_string()))
            .unwrap()
    ).await.unwrap();

    assert_eq!(response.status(), StatusCode::CONFLICT);
}

#[tokio::test]
async fn reorder_changesets_updates_positions() {
    let app = test_app_with_real_mongo().await;
    let (app_id, release_id, cs1, cs2) = seed_draft_release_with_two_changesets(&app).await;

    let response = app.oneshot(
        Request::builder()
            .method("POST")
            .uri(&format!("/api/apps/{app_id}/releases/{release_id}/reorder"))
            .header("content-type", "application/json")
            .header("authorization", config_manager_token())
            .body(Body::from(json!({
                "ordered_changeset_ids": [cs2.to_hex(), cs1.to_hex()]
            }).to_string()))
            .unwrap()
    ).await.unwrap();

    assert_eq!(response.status(), StatusCode::OK);
    let body: serde_json::Value = parse_body(response).await;
    let ids = body["data"]["ordered_changeset_ids"].as_array().unwrap();
    assert_eq!(ids[0].as_str().unwrap(), cs2.to_hex());
    assert_eq!(ids[1].as_str().unwrap(), cs1.to_hex());
}

#[tokio::test]
async fn reorder_rejects_non_permutation() {
    let app = test_app_with_real_mongo().await;
    let (app_id, release_id, cs1, _cs2) = seed_draft_release_with_two_changesets(&app).await;

    let response = app.oneshot(
        Request::builder()
            .method("POST")
            .uri(&format!("/api/apps/{app_id}/releases/{release_id}/reorder"))
            .header("content-type", "application/json")
            .header("authorization", config_manager_token())
            .body(Body::from(json!({
                "ordered_changeset_ids": [cs1.to_hex()]
            }).to_string()))
            .unwrap()
    ).await.unwrap();

    assert_eq!(response.status(), StatusCode::BAD_REQUEST);
}

#[tokio::test]
async fn assembly_worker_composes_changesets_in_order() {
    let ctx = test_worker_context_with_mock_gitaly().await;

    // Mock gitaly: UserMergeToRef succeeds for both changesets.
    ctx.mock_gitaly.expect_user_merge_to_ref()
        .times(2)
        .returning(|_| Ok(UserMergeToRefResponse { commit_id: "abc123".to_string() }));

    let job = seed_assembly_job(&ctx, 2).await;
    let result = execute_release_assemble(&job, &ctx).await;

    assert!(result.is_ok());

    // Verify release is now Validated.
    let release = ctx.release_repo.find_by_id(job.entity_id).await.unwrap();
    assert_eq!(release.state, ReleaseState::Validated);

    // Verify merge SHAs were recorded.
    let rcs = ctx.release_changeset_repo.find_by_release_ordered(release.id).await.unwrap();
    assert!(rcs.iter().all(|rc| rc.merge_sha.is_some()));
}

#[tokio::test]
async fn assembly_worker_marks_changeset_conflicted_on_merge_failure() {
    let ctx = test_worker_context_with_mock_gitaly().await;

    // First merge succeeds, second conflicts.
    ctx.mock_gitaly.expect_user_merge_to_ref()
        .times(1)
        .returning(|_| Ok(UserMergeToRefResponse { commit_id: "abc123".to_string() }));
    ctx.mock_gitaly.expect_user_merge_to_ref()
        .times(1)
        .returning(|_| Err(merge_conflict_error()));

    let job = seed_assembly_job(&ctx, 2).await;
    let result = execute_release_assemble(&job, &ctx).await;

    assert!(result.is_err());

    // Verify release is back to DraftRelease.
    let release = ctx.release_repo.find_by_id(job.entity_id).await.unwrap();
    assert_eq!(release.state, ReleaseState::DraftRelease);

    // Verify the second changeset is marked conflicted.
    let rcs = ctx.release_changeset_repo.find_by_release_ordered(release.id).await.unwrap();
    let cs = ctx.changeset_repo.find_by_id(rcs[1].changeset_id).await.unwrap();
    assert_eq!(cs.state, ChangesetState::Conflicted);
}

#[tokio::test]
async fn publish_creates_tag_and_updates_main() {
    let app = test_app_with_mock_gitaly().await;
    let (app_id, release_id) = seed_validated_release(&app).await;

    // Mock: FindCommit returns current integration branch HEAD.
    app.mock_gitaly.expect_find_commit()
        .returning(|_| Ok(test_git_commit("old_main_sha")));

    // Mock: UserMergeBranch succeeds.
    app.mock_gitaly.expect_user_merge_branch()
        .returning(|_| Ok(test_merge_response("new_sha")));

    // Mock: UserCreateTag succeeds.
    app.mock_gitaly.expect_user_create_tag()
        .returning(|_| Ok(test_tag_response()));

    let response = app.oneshot(
        Request::builder()
            .method("POST")
            .uri(&format!("/api/apps/{app_id}/releases/{release_id}/publish"))
            .header("authorization", config_manager_token())
            .body(Body::empty())
            .unwrap()
    ).await.unwrap();

    assert_eq!(response.status(), StatusCode::OK);

    let body: serde_json::Value = parse_body(response).await;
    assert_eq!(body["data"]["state"], "published");
    assert!(body["data"]["published_sha"].is_string());
    assert!(body["data"]["published_at"].is_string());
}

#[tokio::test]
async fn publish_fails_on_race_condition() {
    let app = test_app_with_mock_gitaly().await;
    let (app_id, release_id) = seed_validated_release(&app).await;

    // Mock: FindCommit returns integration branch HEAD.
    app.mock_gitaly.expect_find_commit()
        .returning(|_| Ok(test_git_commit("old_main_sha")));

    // Mock: UserMergeBranch fails with reference update error (race).
    app.mock_gitaly.expect_user_merge_branch()
        .returning(|_| Err(reference_update_error()));

    let response = app.oneshot(
        Request::builder()
            .method("POST")
            .uri(&format!("/api/apps/{app_id}/releases/{release_id}/publish"))
            .header("authorization", config_manager_token())
            .body(Body::empty())
            .unwrap()
    ).await.unwrap();

    // Race condition surfaces as a conflict error.
    assert_eq!(response.status(), StatusCode::CONFLICT);
}

#[tokio::test]
async fn publish_rejects_non_validated_release() {
    let app = test_app_with_real_mongo().await;
    let (app_id, release_id) = seed_draft_release(&app).await;

    let response = app.oneshot(
        Request::builder()
            .method("POST")
            .uri(&format!("/api/apps/{app_id}/releases/{release_id}/publish"))
            .header("authorization", config_manager_token())
            .body(Body::empty())
            .unwrap()
    ).await.unwrap();

    assert_eq!(response.status(), StatusCode::CONFLICT);
}

#[tokio::test]
async fn published_release_rejects_changeset_modification() {
    let app = test_app_with_real_mongo().await;
    let (app_id, release_id) = seed_published_release(&app).await;

    // Attempt to add a changeset.
    let response = app.oneshot(
        Request::builder()
            .method("POST")
            .uri(&format!("/api/apps/{app_id}/releases/{release_id}/changesets"))
            .header("content-type", "application/json")
            .header("authorization", config_manager_token())
            .body(Body::from(json!({ "add": ["aabbcc"], "remove": [] }).to_string()))
            .unwrap()
    ).await.unwrap();

    assert_eq!(response.status(), StatusCode::CONFLICT);
}

#[tokio::test]
async fn published_release_rejects_reorder() {
    let app = test_app_with_real_mongo().await;
    let (app_id, release_id) = seed_published_release(&app).await;

    let response = app.oneshot(
        Request::builder()
            .method("POST")
            .uri(&format!("/api/apps/{app_id}/releases/{release_id}/reorder"))
            .header("content-type", "application/json")
            .header("authorization", config_manager_token())
            .body(Body::from(json!({ "ordered_changeset_ids": [] }).to_string()))
            .unwrap()
    ).await.unwrap();

    assert_eq!(response.status(), StatusCode::CONFLICT);
}

#[tokio::test]
async fn published_release_rejects_re_assembly() {
    let app = test_app_with_real_mongo().await;
    let (app_id, release_id) = seed_published_release(&app).await;

    let response = app.oneshot(
        Request::builder()
            .method("POST")
            .uri(&format!("/api/apps/{app_id}/releases/{release_id}/assemble"))
            .header("authorization", config_manager_token())
            .body(Body::empty())
            .unwrap()
    ).await.unwrap();

    assert_eq!(response.status(), StatusCode::CONFLICT);
}