const std = @import("std");

pub const PackedState = packed struct(u32) {
    remaining_items: u24 = 0, // Unclaimed entries.
    active_readers: u8 = 0, // Consumers copying descriptors.
};

pub const Config = struct {
    ChunkSize: usize = 256,
};

/// Single-producer, multiple-consumer batch mailbox. Keep do not copy, keep it's address stable.
pub fn CommitQueue(comptime T: type, comptime config: Config) type {
    if (config.ChunkSize == 0 or config.ChunkSize > std.math.maxInt(u24))
        @compileError("ChunkSize must be between 1 and 16777215");
    return struct {
        pub const Item = T;
        pub const ChunkSize = config.ChunkSize;

        // in terms of threadsafety the allocator needs to be coherent to the producer only
        // so a threadlocal arena is pretty ideal for this.
        allocator: std.mem.Allocator,
        state: std.atomic.Value(u32) = .init(0),

        active: std.ArrayList(T) = .empty,
        back: std.ArrayList(T) = .empty,

        pub fn init(allocator: std.mem.Allocator) @This() {
            return .{ .allocator = allocator };
        }

        pub fn deinit(self: *@This()) void {
            while (!self.canSubmit()) std.atomic.spinLoopHint();

            self.active.deinit(self.allocator);
            self.back.deinit(self.allocator);
        }

        /// Producer only
        pub fn push(self: *@This(), item: T) !void {
            if (self.back.items.len == std.math.maxInt(u24)) return error.BatchFull;
            try self.back.append(self.allocator, item);
        }

        /// Copy up to ChunkSize values, caller's responsible for storage.
        pub fn maybeTake(self: *@This(), output: *[ChunkSize]T) []const T {
            var s: PackedState = @bitCast(self.state.load(.monotonic));

            while (true) {
                if (s.remaining_items == 0 or s.active_readers == std.math.maxInt(u8))
                    return &.{};

                // Claim a contiguous chunk and pin its batch atomically.
                var desired = s;
                const count: u24 = @intCast(@min(@as(usize, s.remaining_items), ChunkSize));
                desired.remaining_items -= count;
                desired.active_readers += 1;

                if (self.state.cmpxchgWeak(@bitCast(s), @bitCast(desired), .acquire, .monotonic)) |observed| {
                    s = @bitCast(observed);
                    continue;
                }

                // Read metadata only after successfully pinning.
                const index = self.active.items.len - @as(usize, s.remaining_items);
                @memcpy(output[0..count], self.active.items[index..][0..count]);

                self.releaseReader();
                return output[0..count];
            }
        }

        /// Release a pin after copying the submission.
        fn releaseReader(self: *@This()) void {
            // use shift + fetchsub to avoid a cas loop
            const previous = self.state.fetchSub(@as(u32, 1) << 24, .release);
            std.debug.assert(@as(PackedState, @bitCast(previous)).active_readers != 0);
        }

        pub fn canSubmit(self: *const @This()) bool {
            return self.state.load(.acquire) == 0;
        }

        /// Must only be called by producer thread only.
        /// publishes the backbuffer to all consumers,
        pub fn submit(self: *@This()) void {
            while (!self.canSubmit()) {
                std.atomic.spinLoopHint();
            }

            std.mem.swap(@TypeOf(self.active), &self.active, &self.back);

            self.back.clearRetainingCapacity();
            self.state.store(@bitCast(PackedState{ .remaining_items = @intCast(self.active.items.len) }), .release);
        }
    };
}