kasl_server/

ingest.rs

//! The upload endpoints: `POST /api/v1/days` and `/days/batch`.
//!
//! An agent sends a day (the workday, its pauses, and the tasks recorded on it)
//! and the server writes it whole or not at all. After time offline it sends a
//! stretch of them at once; every day in a batch travels the same path as a
//! live one, and is written on its own so a day that cannot be accepted does
//! not hold up the rest (ADR 0005).
//!
//! Two rules define the contract, both settled before a line of this was
//! written:
//!
//! * **The agent is the source of truth.** A re-upload overwrites what the
//!   server holds for that date. The employee edits their day in kasl - fixes
//!   a task, adds a break they took - and the correction has to land. As a
//!   consequence the same payload sent twice leaves the same rows, which is
//!   what makes a retry after a lost connection safe.
//! * **Timestamps carry an offset, and the day carries its own date.** kasl
//!   stores bare wall-clock text; sending that as-is would make one team's
//!   hours incomparable across time zones. See ADR 0003.

use axum::{Json, extract::State, http::StatusCode, response::IntoResponse};
use chrono::{DateTime, FixedOffset, NaiveDate, Utc};
use serde::{Deserialize, Serialize};
use sqlx::{Postgres, Transaction};
use uuid::Uuid;

use crate::{app::AppState, auth::AuthenticatedAgent, error::ApiError};

/// One day as the agent recorded it.
#[derive(Debug, Deserialize)]
pub struct DayUpload {
    /// The employee's local calendar date, `YYYY-MM-DD`. Sent explicitly
    /// rather than derived from `started_at`: which day work belongs to is the
    /// agent's call, and near midnight the two disagree.
    pub date: NaiveDate,
    /// When the day started, with the agent's UTC offset.
    pub started_at: DateTime<FixedOffset>,
    /// When it ended; absent while the day is still open.
    #[serde(default)]
    pub ended_at: Option<DateTime<FixedOffset>>,
    #[serde(default)]
    pub pauses: Vec<PauseUpload>,
    #[serde(default)]
    pub tasks: Vec<TaskUpload>,
    /// Whether `tasks` is everything the agent holds for this date.
    ///
    /// When set, a task the server stored on this date and the agent no longer
    /// sends is one the employee deleted, and it is removed here too. Tasks on
    /// other dates are untouched - they are matched by id and outlive a single
    /// day, so wiping by date alone would take yesterday's copy with it.
    ///
    /// Defaults to false: an older agent, which cannot know about this flag,
    /// must never have its silence read as "delete the rest".
    #[serde(default)]
    pub tasks_are_complete: bool,
}

#[derive(Debug, Deserialize)]
pub struct PauseUpload {
    pub started_at: DateTime<FixedOffset>,
    #[serde(default)]
    pub ended_at: Option<DateTime<FixedOffset>>,
    /// Seconds. The agent merges neighbouring pauses before sending, so this
    /// is not always `ended_at - started_at` and is taken as given.
    #[serde(default)]
    pub duration_seconds: Option<i32>,
    /// A break the employee entered by hand (the agent's `protected` flag).
    #[serde(default)]
    pub manual: bool,
    #[serde(default)]
    pub reason: Option<String>,
}

#[derive(Debug, Deserialize)]
pub struct TaskUpload {
    /// The agent's own row id. The key a re-upload matches on, so a corrected
    /// task updates instead of piling up.
    pub agent_task_id: i32,
    /// The agent's `task_id`: the same work carried across several days.
    /// Defaults to `agent_task_id`, which is what the agent stores for a task
    /// started today.
    #[serde(default)]
    pub agent_group_id: Option<i32>,
    pub recorded_at: DateTime<FixedOffset>,
    pub name: String,
    #[serde(default)]
    pub comment: Option<String>,
    /// Percent complete, 0..=100.
    pub completeness: i16,
}

/// What the agent gets back: enough to log, and to notice a silent no-op.
#[derive(Debug, Serialize)]
pub struct DayAccepted {
    pub workday_id: Uuid,
    pub date: NaiveDate,
    pub pauses: usize,
    pub tasks: usize,
    /// Tasks dropped because the agent declared its set authoritative. Zero on
    /// the common upload; a non-zero count is worth noticing in a log.
    pub deleted_tasks: u64,
}

/// A stretch of days at once - what an agent sends after time offline.
#[derive(Debug, Deserialize)]
pub struct BatchUpload {
    pub days: Vec<DayUpload>,
}

/// What came of a batch. Counts first: a caller that reads only the status sees
/// `200` even when a day was refused, so the summary has to be impossible to
/// miss in the body.
#[derive(Debug, Serialize)]
pub struct BatchResult {
    pub accepted: usize,
    pub rejected: usize,
    pub results: Vec<DayResult>,
}

/// One day's fate, in the order the days were sent.
#[derive(Debug, Serialize)]
#[serde(tag = "status", rename_all = "lowercase")]
pub enum DayResult {
    Accepted {
        #[serde(flatten)]
        day: DayAccepted,
    },
    /// The date is echoed even here: it is how the agent knows which of its
    /// pending days to keep, and a day can be refused before anything else
    /// about it is known to be usable.
    Rejected { date: NaiveDate, error: String },
}

/// Accepts one day from an authenticated agent.
pub async fn upload_day(State(state): State<AppState>, agent: AuthenticatedAgent, Json(day): Json<DayUpload>) -> Result<impl IntoResponse, ApiError> {
    let accepted = store_day(&state.pool, agent, &day).await?;
    Ok((StatusCode::OK, Json(accepted)))
}

/// Accepts a backlog of days, each written on its own.
///
/// One bad day does not sink the batch. An agent holding a day the server will
/// never accept would otherwise be unable to deliver any of its backlog, and
/// would retry the same doomed request forever (ADR 0005).
pub async fn upload_batch(State(state): State<AppState>, agent: AuthenticatedAgent, Json(batch): Json<BatchUpload>) -> Result<impl IntoResponse, ApiError> {
    if batch.days.len() > state.max_batch_days {
        return Err(ApiError::new(
            StatusCode::PAYLOAD_TOO_LARGE,
            format!("a batch carries at most {} days; split the backlog", state.max_batch_days),
        ));
    }

    let mut results = Vec::with_capacity(batch.days.len());
    let mut accepted = 0;
    let mut rejected = 0;

    for day in &batch.days {
        match store_day(&state.pool, agent, day).await {
            Ok(stored) => {
                accepted += 1;
                results.push(DayResult::Accepted { day: stored });
            }
            // A day the server itself failed on aborts the batch: the agent
            // must retry it, and reporting a database outage as "this day is
            // rejected" would tell it to give up instead.
            Err(error) if error.status().is_server_error() => return Err(error),
            Err(error) => {
                rejected += 1;
                results.push(DayResult::Rejected {
                    date: day.date,
                    error: error.to_string(),
                });
            }
        }
    }

    tracing::info!(user_id = %agent.user_id, agent_id = %agent.agent_id, accepted, rejected, "accepted a batch");

    Ok((StatusCode::OK, Json(BatchResult { accepted, rejected, results })))
}

/// Writes one day, whole or not at all.
///
/// Shared by the single-day route and the batch one so a backfilled day is
/// stored by exactly the same code as a live one.
async fn store_day(pool: &sqlx::PgPool, agent: AuthenticatedAgent, day: &DayUpload) -> Result<DayAccepted, ApiError> {
    validate(day)?;

    // All of it or none: a day whose pauses landed but whose tasks did not
    // would show up on a dashboard as real, and nobody would know to re-send.
    let mut tx = pool.begin().await?;

    let workday_id = upsert_workday(&mut tx, agent.user_id, day).await?;
    replace_pauses(&mut tx, workday_id, &day.pauses).await?;
    let tasks = upsert_tasks(&mut tx, agent.user_id, day.date, &day.tasks).await?;
    let deleted_tasks = if day.tasks_are_complete {
        delete_missing_tasks(&mut tx, agent.user_id, day.date, &day.tasks).await?
    } else {
        0
    };

    tx.commit().await?;

    // The agent, not just the person: several machines report for one employee
    // and "which one sent this" is the first question when a day looks wrong.
    tracing::info!(%workday_id, user_id = %agent.user_id, agent_id = %agent.agent_id, date = %day.date, pauses = day.pauses.len(), tasks, deleted_tasks, "accepted a day");

    Ok(DayAccepted {
        workday_id,
        date: day.date,
        pauses: day.pauses.len(),
        tasks,
        deleted_tasks,
    })
}

/// Rejects payloads the schema would refuse anyway, with a message that says
/// which field is wrong - a constraint violation surfaces as a 500 and tells
/// the agent nothing it can act on.
fn validate(day: &DayUpload) -> Result<(), ApiError> {
    if let Some(ended_at) = day.ended_at
        && ended_at < day.started_at
    {
        return Err(ApiError::bad_request("ended_at is before started_at"));
    }

    for (index, pause) in day.pauses.iter().enumerate() {
        if let Some(ended_at) = pause.ended_at
            && ended_at < pause.started_at
        {
            return Err(ApiError::bad_request(format!("pauses[{index}]: ended_at is before started_at")));
        }
        if pause.duration_seconds.is_some_and(|seconds| seconds < 0) {
            return Err(ApiError::bad_request(format!("pauses[{index}]: duration_seconds is negative")));
        }
    }

    for (index, task) in day.tasks.iter().enumerate() {
        if !(0..=100).contains(&task.completeness) {
            return Err(ApiError::bad_request(format!("tasks[{index}]: completeness must be between 0 and 100")));
        }
        if task.name.trim().is_empty() {
            return Err(ApiError::bad_request(format!("tasks[{index}]: name is empty")));
        }
    }

    Ok(())
}

/// Writes the day, or corrects the one already stored for that date.
async fn upsert_workday(tx: &mut Transaction<'_, Postgres>, user_id: Uuid, day: &DayUpload) -> Result<Uuid, ApiError> {
    let workday_id: Uuid = sqlx::query_scalar(
        "INSERT INTO workdays (user_id, date, started_at, ended_at) VALUES ($1, $2, $3, $4)
         ON CONFLICT (user_id, date) DO UPDATE SET started_at = EXCLUDED.started_at, ended_at = EXCLUDED.ended_at
         RETURNING id",
    )
    .bind(user_id)
    .bind(day.date)
    .bind(day.started_at.with_timezone(&Utc))
    .bind(day.ended_at.map(|at| at.with_timezone(&Utc)))
    .fetch_one(&mut **tx)
    .await?;

    Ok(workday_id)
}

/// Replaces the day's pauses wholesale.
///
/// Pauses have no agent-side identity to match on - the agent splits and
/// merges them as activity comes in - so the day's set is what was sent, and
/// a pause the employee deleted disappears here too.
async fn replace_pauses(tx: &mut Transaction<'_, Postgres>, workday_id: Uuid, pauses: &[PauseUpload]) -> Result<(), ApiError> {
    sqlx::query("DELETE FROM pauses WHERE workday_id = $1")
        .bind(workday_id)
        .execute(&mut **tx)
        .await?;

    for pause in pauses {
        sqlx::query("INSERT INTO pauses (workday_id, started_at, ended_at, duration_seconds, manual, reason) VALUES ($1, $2, $3, $4, $5, $6)")
            .bind(workday_id)
            .bind(pause.started_at.with_timezone(&Utc))
            .bind(pause.ended_at.map(|at| at.with_timezone(&Utc)))
            .bind(pause.duration_seconds)
            .bind(pause.manual)
            .bind(pause.reason.as_deref())
            .execute(&mut **tx)
            .await?;
    }

    Ok(())
}

/// Writes the day's tasks, correcting any the agent has sent before.
///
/// Tasks do carry an agent-side id, so they are matched rather than replaced:
/// the same task may appear on several days, and wiping by date would take
/// yesterday's copy with it.
async fn upsert_tasks(tx: &mut Transaction<'_, Postgres>, user_id: Uuid, date: NaiveDate, tasks: &[TaskUpload]) -> Result<usize, ApiError> {
    for task in tasks {
        sqlx::query(
            "INSERT INTO tasks (user_id, agent_task_id, agent_group_id, date, recorded_at, name, comment, completeness)
             VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
             ON CONFLICT (user_id, agent_task_id) DO UPDATE SET
                 agent_group_id = EXCLUDED.agent_group_id,
                 date = EXCLUDED.date,
                 recorded_at = EXCLUDED.recorded_at,
                 name = EXCLUDED.name,
                 comment = EXCLUDED.comment,
                 completeness = EXCLUDED.completeness",
        )
        .bind(user_id)
        .bind(task.agent_task_id)
        .bind(task.agent_group_id.unwrap_or(task.agent_task_id))
        .bind(date)
        .bind(task.recorded_at.with_timezone(&Utc))
        .bind(task.name.trim())
        .bind(task.comment.as_deref())
        .bind(task.completeness)
        .execute(&mut **tx)
        .await?;
    }

    Ok(tasks.len())
}

/// Removes the date's tasks the agent did not send.
///
/// Only reached when the agent marked its list authoritative. Scoped to the
/// one date on purpose: a task carried across several days keeps its rows on
/// the others, and an agent backfilling Monday cannot erase Friday.
async fn delete_missing_tasks(tx: &mut Transaction<'_, Postgres>, user_id: Uuid, date: NaiveDate, tasks: &[TaskUpload]) -> Result<u64, ApiError> {
    let kept: Vec<i32> = tasks.iter().map(|task| task.agent_task_id).collect();

    let deleted = sqlx::query("DELETE FROM tasks WHERE user_id = $1 AND date = $2 AND agent_task_id <> ALL($3)")
        .bind(user_id)
        .bind(date)
        .bind(&kept)
        .execute(&mut **tx)
        .await?
        .rows_affected();

    Ok(deleted)
}

#[cfg(test)]
mod tests {
    use super::*;

    fn day_json(patch: serde_json::Value) -> DayUpload {
        let mut value = serde_json::json!({
            "date": "2026-08-14",
            "started_at": "2026-08-14T09:00:00-03:00",
            "pauses": [],
            "tasks": [],
        });
        let (serde_json::Value::Object(base), serde_json::Value::Object(patch)) = (&mut value, patch) else {
            panic!("both must be objects");
        };
        base.extend(patch);
        serde_json::from_value(value).expect("the fixture should deserialize")
    }

    #[test]
    fn an_offset_is_required_on_every_instant() {
        // The whole point of the contract: bare wall-clock time, which is what
        // kasl stores locally, must not parse.
        let bare = serde_json::json!({
            "date": "2026-08-14",
            "started_at": "2026-08-14T09:00:00",
        });
        assert!(
            serde_json::from_value::<DayUpload>(bare).is_err(),
            "an instant without an offset must be rejected"
        );
    }

    #[test]
    fn the_offset_is_preserved_as_an_instant() {
        let day = day_json(serde_json::json!({ "started_at": "2026-08-14T09:00:00-03:00" }));
        assert_eq!(day.started_at.with_timezone(&Utc).to_rfc3339(), "2026-08-14T12:00:00+00:00");
    }

    #[test]
    fn a_day_may_still_be_open() {
        let day = day_json(serde_json::json!({}));
        assert!(day.ended_at.is_none(), "a missing ended_at means the day is still running");
        validate(&day).expect("an open day is valid");
    }

    #[test]
    fn a_day_cannot_end_before_it_starts() {
        let day = day_json(serde_json::json!({ "ended_at": "2026-08-14T08:00:00-03:00" }));
        let error = validate(&day).expect_err("a backwards day must be refused");
        assert_eq!(error.to_string(), "ended_at is before started_at");
    }

    #[test]
    fn impossible_pauses_and_tasks_are_named_in_the_error() {
        let day = day_json(serde_json::json!({
            "pauses": [
                {"started_at": "2026-08-14T10:00:00-03:00", "ended_at": "2026-08-14T10:20:00-03:00", "duration_seconds": 1200},
                {"started_at": "2026-08-14T12:00:00-03:00", "duration_seconds": -1},
            ],
        }));
        let error = validate(&day).expect_err("a negative duration must be refused");
        assert!(
            error.to_string().contains("pauses[1]"),
            "the message should point at the offending element: {error}"
        );

        let day = day_json(serde_json::json!({
            "tasks": [{"agent_task_id": 1, "recorded_at": "2026-08-14T17:00:00-03:00", "name": "Ship it", "completeness": 101}],
        }));
        let error = validate(&day).expect_err("completeness above 100 must be refused");
        assert!(
            error.to_string().contains("tasks[0]"),
            "the message should point at the offending element: {error}"
        );
    }

    #[test]
    fn a_task_group_defaults_to_the_task_itself() {
        let day = day_json(serde_json::json!({
            "tasks": [{"agent_task_id": 7, "recorded_at": "2026-08-14T17:00:00-03:00", "name": "Write the ingest", "completeness": 60}],
        }));
        let task = &day.tasks[0];
        assert_eq!(task.agent_group_id, None, "an absent group is absent on the wire");
        assert_eq!(task.agent_group_id.unwrap_or(task.agent_task_id), 7, "and resolves to the task itself");
    }

    #[test]
    fn an_agent_that_says_nothing_deletes_nothing() {
        // The compatibility hinge: agents shipped before this flag existed send
        // whatever tasks they have, and their silence must not be read as
        // "delete everything else on that date".
        let day = day_json(serde_json::json!({}));
        assert!(!day.tasks_are_complete, "the authoritative set must be opt-in");

        let day = day_json(serde_json::json!({ "tasks_are_complete": true }));
        assert!(day.tasks_are_complete, "and an agent that opts in is heard");
    }

    #[test]
    fn a_nameless_task_is_refused() {
        let day = day_json(serde_json::json!({
            "tasks": [{"agent_task_id": 1, "recorded_at": "2026-08-14T17:00:00-03:00", "name": "   ", "completeness": 50}],
        }));
        assert!(validate(&day).is_err(), "a task with a blank name carries no information");
    }
}