const std = @import("std");

const Io = std.Io;

var store_io: Io = undefined;
var initialized = false;
var mutex: Io.Mutex = .init;
var changed: Io.Condition = .init;
var generation: usize = 0;
var latest_seq: u64 = 0;

pub fn init(io: Io) void {
    store_io = io;
    initialized = true;
}

pub fn publish(seq: u64) void {
    if (!initialized) return;

    mutex.lockUncancelable(store_io);
    defer mutex.unlock(store_io);

    _ = @atomicRmw(usize, &generation, .Add, 1, .release);
    latest_seq = @max(latest_seq, seq);
    changed.broadcast(store_io);
}

pub fn currentGeneration() usize {
    return @atomicLoad(usize, &generation, .acquire);
}

fn snapshotLocked() Snapshot {
    return .{
        .generation = currentGeneration(),
        .latest_seq = latest_seq,
    };
}

pub fn waitForChange(observed: usize, stopped: *const bool) !?Snapshot {
    try mutex.lock(store_io);
    defer mutex.unlock(store_io);

    while (@atomicLoad(usize, &generation, .acquire) == observed and
        !@atomicLoad(bool, stopped, .acquire))
    {
        try changed.wait(store_io, &mutex);
    }
    if (@atomicLoad(bool, stopped, .acquire)) return null;
    return snapshotLocked();
}

pub fn wakeWaiters() void {
    if (!initialized) return;

    mutex.lockUncancelable(store_io);
    defer mutex.unlock(store_io);
    changed.broadcast(store_io);
}

pub const Snapshot = struct {
    generation: usize,
    latest_seq: u64,
};

test "wakeWaiters releases a stopped idle waiter" {
    init(std.Options.debug_io);

    var stopped = false;
    var returned = false;
    const observed = currentGeneration();
    const thread = try std.Thread.spawn(.{}, struct {
        fn run(seen: usize, stop: *bool, done: *bool) void {
            const result = waitForChange(seen, stop) catch unreachable;
            std.debug.assert(result == null);
            @atomicStore(bool, done, true, .release);
        }
    }.run, .{ observed, &stopped, &returned });

    std.Options.debug_io.sleep(.fromMilliseconds(10), .awake) catch unreachable;
    @atomicStore(bool, &stopped, true, .release);
    wakeWaiters();
    thread.join();

    try std.testing.expect(@atomicLoad(bool, &returned, .acquire));
}