zcbot/windows-node/Zcbot.WindowsNode/Jobs/JobRepository.cs

using System.Text.Json;

namespace Zcbot.WindowsNode;

internal sealed class JobRepository(string jobsDirectory)
{
    private string JobsDirectory => jobsDirectory;

    internal string JobDirectory(Guid jobId) =>
        PathGuard.CombineUnderRoot(JobsDirectory, jobId.ToString("D"));

    internal bool HasPendingJobs => Directory.Exists(JobsDirectory)
        && ReadRecoverableJobs().Any(item => item.Terminal is null);

    internal IReadOnlyList<RecoverableJob> ReadRecoverableJobs()
    {
        if (!Directory.Exists(JobsDirectory))
        {
            return [];
        }
        var jobs = new List<RecoverableJob>();
        foreach (var requestPath in Directory.EnumerateFiles(
            JobsDirectory, "request.json", SearchOption.AllDirectories))
        {
            try
            {
                using var request = JsonDocument.Parse(File.ReadAllBytes(requestPath));
                var root = request.RootElement;
                if (!JobInboxStore.TryReadGuid(root, "job_id", out var jobId)
                    || !JobInboxStore.TryReadGuid(root, "lease_id", out var leaseId)
                    || !root.TryGetProperty("request_digest", out var digestValue)
                    || digestValue.GetString() is not { Length: 64 } requestDigest
                    || !root.TryGetProperty("capability", out var capabilityValue)
                    || capabilityValue.GetString() is not { Length: > 0 } capability
                    || !root.TryGetProperty("workspace", out var workspaceValue)
                    || !JobInboxStore.TryReadWorkspace(workspaceValue, out var workspace))
                {
                    continue;
                }
                var jobDirectory = Directory.GetParent(
                    Directory.GetParent(requestPath)!.FullName)!.FullName;
                jobs.Add(new RecoverableJob(
                    jobId,
                    leaseId,
                    requestDigest,
                    capability,
                    workspace,
                    root.TryGetProperty("input_transfers", out var transfers)
                        ? transfers.Clone() : null,
                    JobInboxStore.ReadTerminal(Path.Combine(jobDirectory, "terminal.json")),
                    File.Exists(Path.Combine(jobDirectory, "upload-complete.json")),
                    File.Exists(Path.Combine(jobDirectory, "cloud-terminal.json"))));
            }
            catch (Exception exception) when (
                exception is JsonException or IOException or UnauthorizedAccessException)
            {
            }
        }
        return jobs;
    }

    internal void WriteState(
        RecoverableJob job,
        string stage,
        int progress,
        string detail,
        JobTransitionMode mode = JobTransitionMode.Normal)
    {
        var path = Path.Combine(JobDirectory(job.JobId), "state.json");
        var current = JobInboxStore.ReadState(path)?.Stage ?? JobStateMachine.Accepted;
        JobStateMachine.EnsureTransition(current, stage, mode);
        var content = JsonSerializer.SerializeToUtf8Bytes(new
        {
            stage,
            progress = Math.Clamp(progress, 0, 100),
            detail = detail[..Math.Min(detail.Length, 500)],
            updated_at = DateTimeOffset.UtcNow,
        }, JobInboxStore.JsonOptions);
        try
        {
            AtomicFile.Write(path, content, overwrite: true);
        }
        catch (Exception exception) when (
            exception is IOException or UnauthorizedAccessException)
        {
            Console.Error.WriteLine($"[WARN] Local job state update failed: {exception.Message}");
        }
    }

    internal void EnsureAcceptedState(Guid jobId)
    {
        var path = Path.Combine(JobDirectory(jobId), "state.json");
        if (File.Exists(path))
        {
            return;
        }
        var content = JsonSerializer.SerializeToUtf8Bytes(new
        {
            stage = JobStateMachine.Accepted,
            progress = 0,
            detail = "任务已由本机接收",
            updated_at = DateTimeOffset.UtcNow,
        }, JobInboxStore.JsonOptions);
        try
        {
            AtomicFile.Write(path, content, overwrite: false);
        }
        catch (IOException) when (File.Exists(path))
        {
        }
        catch (Exception exception) when (
            exception is IOException or UnauthorizedAccessException)
        {
            Console.Error.WriteLine($"[WARN] Initial local job state failed: {exception.Message}");
        }
    }

    internal IReadOnlyList<JsonElement> InputTransfers(RecoverableJob job)
    {
        if (job.InputTransfers is not JsonElement transfers
            || !JobInboxStore.IsValidInputTransfers(transfers, job.JobId))
        {
            throw new InvalidDataException("Stored input transfers are invalid.");
        }
        return transfers.EnumerateArray().Select(item => item.Clone()).ToArray();
    }

    internal string InputPath(RecoverableJob job, JsonElement transfer)
    {
        var key = transfer.GetProperty("key").GetString()!;
        var filename = transfer.GetProperty("filename").GetString()!;
        return PathGuard.CombineUnderRoot(
            JobDirectory(job.JobId), "input", key, filename);
    }

    internal void WriteTerminal(
        RecoverableJob job,
        string status,
        string code,
        string detail,
        bool overwrite = false)
    {
        JobStateMachine.EnsureTerminalStatus(status);
        var path = Path.Combine(JobDirectory(job.JobId), "terminal.json");
        if (File.Exists(path) && !overwrite)
        {
            return;
        }
        var content = JsonSerializer.SerializeToUtf8Bytes(new
        {
            job_id = job.JobId,
            lease_id = job.LeaseId,
            request_digest = job.RequestDigest,
            status,
            error = new { code, detail },
            artifact_manifest = Array.Empty<object>(),
            terminal_at = DateTimeOffset.UtcNow,
        }, JobInboxStore.JsonOptions);
        AtomicFile.Write(path, content, overwrite);
    }
}