VolcanicAgenticCore/extensions/Volcanic.AgenticCore.Extensions.Scheduler/JobStore.cs

using System.Globalization;
using System.Text.Json;
using System.Text.Json.Nodes;
using Microsoft.Extensions.Logging;
using Volcanic.AgenticCore.Abstractions.Json;

namespace Volcanic.AgenticCore.Extensions.Scheduler;

/// <summary>
/// <c>jobs.json</c>: <c>{ "version": 1, "jobs": [...] }</c> in <see cref="VolcanicJson"/> camelCase. Written atomically
/// (temp file + rename) through one semaphore. A corrupt file is set aside and loading starts empty, so a bad file never
/// keeps the host down; invalid jobs (including entries that don't deserialise) are dropped with a warning. A job of the
/// retired <c>reminder</c> kind is loaded as a prompt asking the agent to pass the reminder on.
/// <para>
/// A chat's own file (<see cref="PathIn"/>, with <paramref name="target"/>) owns its jobs: every entry gets that target,
/// whatever it stores. A VolcanicSecretary <c>jobs.json</c> there (a bare array without targets) is converted, saved in
/// this format and the original kept as <c>jobs.json.secretary-&lt;timestamp&gt;</c>. Without a target (the retired
/// shared file) entries keep their own targets and a VolcanicSecretary file is set aside.
/// </para>
/// </summary>
public sealed class JobStore(string path, TimeProvider time, ILogger logger, string? target = null)
{
    public const int Version = 1;
    private const string ReminderPrefix = "Remind me: ";
    private readonly SemaphoreSlim _writeGate = new(1, 1);

    public string Path => path;

    /// <summary>A chat's job file: <c>&lt;workspace&gt;/.volcanic/scheduler/jobs.json</c>.</summary>
    public static string PathIn(string workspace) => System.IO.Path.Combine(workspace, ".volcanic", "scheduler", "jobs.json");

    public async Task<IReadOnlyList<ScheduledJob>> LoadAsync(CancellationToken cancellationToken)
    {
        if (!File.Exists(path)) return [];
        var json = await File.ReadAllTextAsync(path, cancellationToken).ConfigureAwait(false);
        JsonElement root;
        try
        {
            using var document = JsonDocument.Parse(json);
            root = document.RootElement.Clone();
        }
        catch (JsonException)
        {
            root = default;
        }
        if (root.ValueKind == JsonValueKind.Array)
            return await LoadSecretaryAsync(root, cancellationToken).ConfigureAwait(false);

        StoredFile? file;
        try
        {
            file = root.ValueKind == JsonValueKind.Object ? VolcanicJson.Deserialize<StoredFile>(root) : null;
        }
        catch (JsonException)
        {
            file = null;
        }
        if (file is not { Version: Version, Jobs: not null })
        {
            SetAside("could not be read");
            return [];
        }
        return ReadAll([.. file.Jobs.Select(Owned)]);
    }

    private async Task<IReadOnlyList<ScheduledJob>> LoadSecretaryAsync(JsonElement entries, CancellationToken cancellationToken)
    {
        if (string.IsNullOrWhiteSpace(target))
        {
            SetAside("is a VolcanicSecretary job file, which belongs in a chat's workspace (.volcanic/scheduler/jobs.json)");
            return [];
        }
        var source = entries.EnumerateArray().ToList();
        var jobs = ReadAll([.. source.Select(FromSecretary)]);
        var backup = $"{path}.secretary-{Stamp()}";
        File.Copy(path, backup, overwrite: true);
        await SaveAsync(jobs, cancellationToken).ConfigureAwait(false);
        logger.LogInformation("Converted the VolcanicSecretary job file {Path} for target {Target}: kept {Kept} of {Total} jobs; the original is {Backup}",
            path, target, jobs.Count, source.Count, backup);
        return jobs;
    }

    private List<ScheduledJob> ReadAll(IReadOnlyList<JsonElement> entries)
    {
        var jobs = new List<ScheduledJob>();
        var ids = new HashSet<string>(StringComparer.Ordinal);
        for (var i = 0; i < entries.Count; i++)
        {
            var (job, problem) = Read(entries[i]);
            problem ??= ids.Add(job!.Id) ? null : "duplicate id";
            if (problem is not null)
            {
                logger.LogWarning("Dropped scheduled job #{Index} ({Id}) from {Path}: {Problem}", i, job?.Id, path, problem);
                continue;
            }
            jobs.Add(job!);
        }
        return jobs;
    }

    private void SetAside(string why)
    {
        var aside = $"{path}.corrupt-{Stamp()}";
        File.Move(path, aside, overwrite: true);
        logger.LogError("The job store {Path} {Why}; it was moved to {Aside} and the scheduler starts empty", path, why, aside);
    }

    private string Stamp() => time.GetUtcNow().ToString("yyyyMMddHHmmss", CultureInfo.InvariantCulture);

    public async Task SaveAsync(IReadOnlyList<ScheduledJob> jobs, CancellationToken cancellationToken)
    {
        await _writeGate.WaitAsync(cancellationToken).ConfigureAwait(false);
        var temp = $"{path}.{Environment.ProcessId.ToString(CultureInfo.InvariantCulture)}.tmp";
        try
        {
            Directory.CreateDirectory(System.IO.Path.GetDirectoryName(System.IO.Path.GetFullPath(path))!);
            var json = JsonSerializer.Serialize(new JobFile(Version, [.. jobs]), VolcanicJson.Indented);
            await File.WriteAllTextAsync(temp, json, cancellationToken).ConfigureAwait(false);
            File.Move(temp, path, overwrite: true);
        }
        catch
        {
            try { File.Delete(temp); } catch (IOException) { } catch (UnauthorizedAccessException) { }
            throw;
        }
        finally
        {
            _writeGate.Release();
        }
    }

    /// <summary>One entry of the file: the job, or why it's dropped. An entry that doesn't deserialise only loses itself.</summary>
    private static (ScheduledJob? Job, string? Problem) Read(JsonElement entry)
    {
        if (entry.ValueKind != JsonValueKind.Object) return (null, "not a JSON object");
        ScheduledJob? job;
        try
        {
            job = VolcanicJson.Deserialize<ScheduledJob>(FromReminder(entry));
        }
        catch (JsonException ex)
        {
            return (null, ex.Message);
        }
        if (job is null) return (null, "empty entry");
        return (job, JobValidation.CheckStored(job));
    }

    /// <summary>In a chat's file the entry belongs to that chat, whatever target it stores.</summary>
    private JsonElement Owned(JsonElement entry)
    {
        if (string.IsNullOrWhiteSpace(target) || entry.ValueKind != JsonValueKind.Object) return entry;
        var node = JsonNode.Parse(entry.GetRawText())!.AsObject();
        node["target"] = target;
        return JsonSerializer.SerializeToElement(node);
    }

    /// <summary>A stored <c>reminder</c> (fixed text sent as is) becomes a prompt; other entries pass through unchanged.</summary>
    private static JsonElement FromReminder(JsonElement entry)
    {
        if (!entry.TryGetProperty("kind", out var kind) || kind.ValueKind != JsonValueKind.String
            || !string.Equals(kind.GetString(), "reminder", StringComparison.OrdinalIgnoreCase))
            return entry;
        var node = JsonNode.Parse(entry.GetRawText())!.AsObject();
        node["kind"] = "prompt";
        if (entry.TryGetProperty("text", out var text) && Remind(text) is { } reminder) node["text"] = reminder;
        return JsonSerializer.SerializeToElement(node);
    }

    /// <summary>The text of a reminder with <see cref="ReminderPrefix"/>, or null when it's empty or the prefix doesn't fit.</summary>
    private static string? Remind(JsonElement text) =>
        text.ValueKind == JsonValueKind.String && text.GetString() is { } body && !string.IsNullOrWhiteSpace(body)
        && body.Length + ReminderPrefix.Length <= JobValidation.MaxTextLength
            ? ReminderPrefix + body
            : null;

    /// <summary>
    /// One VolcanicSecretary job as an entry of this format. <c>source.kind</c> <c>text</c> becomes a prompt (a
    /// <c>deliver: direct</c> one, which sent its text as is, a reminder), <c>script</c> a watcher; a 6-field cron with
    /// seconds <c>0</c> loses that field. Anything else is left for <see cref="Read"/> to reject.
    /// </summary>
    private JsonElement FromSecretary(JsonElement entry)
    {
        if (entry.ValueKind != JsonValueKind.Object) return entry;
        var node = new JsonObject { ["target"] = target };
        foreach (var name in (string[])["id", "name", "createdAt", "lastRunAt"])
            if (entry.TryGetProperty(name, out var value)) node[name] = JsonNode.Parse(value.GetRawText());

        if (entry.TryGetProperty("schedule", out var schedule))
        {
            var copy = JsonNode.Parse(schedule.GetRawText());
            if (copy is JsonObject scheduleNode && scheduleNode["cron"] is JsonValue cron && cron.TryGetValue<string>(out var expression))
                scheduleNode["cron"] = WithoutZeroSeconds(expression);
            node["schedule"] = copy;
        }

        if (entry.TryGetProperty("source", out var source) && source.ValueKind == JsonValueKind.Object
            && source.TryGetProperty("kind", out var kind) && kind.ValueKind == JsonValueKind.String)
        {
            switch (kind.GetString())
            {
                case "text":
                    node["kind"] = "prompt";
                    if (source.TryGetProperty("text", out var text))
                    {
                        var direct = !entry.TryGetProperty("deliver", out var deliver) || deliver.ValueKind != JsonValueKind.String || deliver.GetString() != "agent";
                        node["text"] = (direct ? Remind(text) : null) ?? JsonNode.Parse(text.GetRawText());
                    }
                    break;
                case "script":
                    node["kind"] = "watcher";
                    if (source.TryGetProperty("command", out var command)) node["command"] = JsonNode.Parse(command.GetRawText());
                    if (source.TryGetProperty("timeoutSec", out var timeout)) node["timeoutSeconds"] = JsonNode.Parse(timeout.GetRawText());
                    break;
                default:
                    node["kind"] = kind.GetString();
                    break;
            }
        }

        if (entry.TryGetProperty("lastResult", out var result) && JsonNode.Parse(result.GetRawText()) is JsonObject resultNode)
        {
            resultNode["timedOut"] ??= false;
            node["lastResult"] = resultNode;
        }
        return JsonSerializer.SerializeToElement(node);
    }

    private static string WithoutZeroSeconds(string cron)
    {
        var fields = cron.Split((char[]?)null, StringSplitOptions.RemoveEmptyEntries);
        return fields.Length == 6 && fields[0] == "0" ? string.Join(' ', fields[1..]) : cron;
    }

    private sealed record JobFile(int Version, List<ScheduledJob> Jobs);

    private sealed record StoredFile(int Version, List<JsonElement>? Jobs);
}