add refresh POST operation to allow forced symbol refreshes

src/App.zig

@@ -35,6 +35,17 @@ pub const App = struct {
     /// Without it two simultaneous adds could both read the old file and
     /// the second writer would clobber the first's new symbol.
     watch_mutex: std.Io.Mutex = .init,
+    /// Serializes `POST /refresh`, and does so with a NON-blocking acquire so a
+    /// concurrent request is rejected (409) rather than queued.
+    ///
+    /// Queueing would be worse than refusing: each refresh spends provider quota,
+    /// the request is synchronous, and zfin's HTTP client retries up to three
+    /// times on a 5xx or transient failure. A refresh slow enough to time out
+    /// would therefore be retried while the first was still running, and a
+    /// blocking lock would dutifully run every one of them in turn - turning one
+    /// slow request into a provider stampede. Refusing tells the caller the truth:
+    /// a refresh is already in flight, so wait for it.
+    refresh_mutex: std.Io.Mutex = .init,
 
     pub fn init(io: std.Io, allocator: std.mem.Allocator, environ: *const std.process.Environ.Map) App {
         const config = zfin.Config.fromEnv(io, allocator, environ);
@@ -218,6 +229,11 @@ test "pathIsPublic: gated surface (key required)" {
     // the cost of that assumption being wrong is silent disclosure.
     try std.testing.expect(!pathIsPublic("/AAPL/diagnostics"));
     try std.testing.expect(!pathIsPublic("/BRK.B/diagnostics"));
+    // `/refresh` spends provider quota and writes the cache. It needs no entry
+    // here to be gated - the default is closed - and this asserts the default
+    // rather than trusting it, because an unauthenticated refresh trigger is a
+    // free denial-of-quota for anyone who finds the URL.
+    try std.testing.expect(!pathIsPublic("/refresh"));
 }
 
 test "pathIsPublic: returns look-alikes do not slip through" {

src/handlers.zig

@@ -84,6 +84,9 @@ pub fn handleHelp(_: *App, _: *httpz.Request, res: *httpz.Response) !void {
         \\  GET /{SYMBOL}/quote             Latest quote (JSON)
         \\  GET /{SYMBOL}/candles           Raw SRF cache file
         \\  GET /{SYMBOL}/candles_meta      Candle freshness metadata (SRF)
+        \\  GET /{SYMBOL}/diagnostics       Whether the server TRACKS the symbol,
+        \\                                  how far behind its peers it is, and
+        \\                                  whether its copy is stamped fresh (JSON)
         \\  GET /{SYMBOL}/dividends         Raw SRF cache file
         \\  GET /{SYMBOL}/splits            Raw SRF cache file
         \\  GET /{SYMBOL}/earnings          Raw SRF cache file
@@ -94,6 +97,8 @@ pub fn handleHelp(_: *App, _: *httpz.Request, res: *httpz.Response) !void {
         \\  GET /_edgar/tickers_funds       EDGAR mutual-fund ticker map (SRF)
         \\  GET /_edgar/tickers_companies   EDGAR company ticker map (SRF)
         \\  GET /symbols                    List of tracked symbols
+        \\  POST /refresh?symbols=A,B       Force-refresh candles now (max 25;
+        \\                                  409 if a refresh is already running)
         \\
         \\Auth:
         \\  All endpoints except /, /help, and /{SYMBOL}/returns require an
@@ -105,6 +110,13 @@ pub fn handleHelp(_: *App, _: *httpz.Request, res: *httpz.Response) !void {
         \\  fetches once from the provider, fills the cache, then serves
         \\  (404 only if that fetch also fails).
         \\
+        \\Freshness (why POST for one and GET for the other):
+        \\  A read never triggers a refetch - a present file is served as-is and
+        \\  cron is the freshness authority. POST /refresh is the one way to make
+        \\  the server fetch on demand, and it is a POST because it spends provider
+        \\  quota and writes the cache. /{SYMBOL}/watch is a GET only because
+        \\  LibreOffice's WEBSERVICE cannot issue anything else.
+        \\
         \\Watchlist (add a symbol to the cron refresh set):
         \\  GET /{SYMBOL}/watch              authenticated; for your own tooling
         \\  GET /{SYMBOL}/returns?watch=true public, but only LibreOffice's
@@ -518,6 +530,147 @@ pub fn handleDiagnostics(app: *App, req: *httpz.Request, res: *httpz.Response) !
     res.body = aw.written();
 }
 
+/// Cap on symbols per `POST /refresh`.
+///
+/// Each one costs a forced provider fetch, so an unbounded list is an unbounded
+/// synchronous request. Generous for the intended use - one or two symbols
+/// breaking a reconciliation - while keeping any single request bounded. Exceeding
+/// it is a 400 naming the limit, never a silent truncation: quietly refreshing the
+/// first 25 of 40 would report success for symbols nobody touched.
+pub const max_refresh_symbols: usize = 25;
+
+/// Outcome of parsing a `symbols=` value. A union rather than an error set so the
+/// 400 can name the offending token - "Invalid symbol" without saying which is a
+/// worse message than the operator's own typo.
+pub const SymbolList = union(enum) {
+    ok: []const []const u8,
+    /// No usable token at all.
+    empty,
+    /// Distinct symbol count after dedupe.
+    too_many: usize,
+    /// The token that failed `isPlausibleSymbol`.
+    malformed: []const u8,
+};
+
+/// Parse a comma-separated `symbols=` value into a deduped, upper-cased list.
+///
+/// Empty tokens are skipped rather than rejected, so a trailing comma is harmless
+/// - punctuation is not a typo'd ticker, and there is nothing an operator could
+/// mean by it. A genuinely malformed token IS rejected, and rejects the whole
+/// request: refreshing the valid remainder would report overall success while the
+/// symbol the operator actually cared about was never touched.
+///
+/// Deduping happens BEFORE the cap check, so `AAPL,AAPL,...` cannot trip a limit
+/// it does not really exceed.
+pub fn parseSymbolList(arena: std.mem.Allocator, raw: []const u8) !SymbolList {
+    var out = std.ArrayList([]const u8).empty;
+    var seen = std.StringHashMap(void).init(arena);
+    var it = std.mem.splitScalar(u8, raw, ',');
+    while (it.next()) |tok| {
+        const trimmed = std.mem.trim(u8, tok, " \t\r\n");
+        if (trimmed.len == 0) continue;
+        const sym = try upperDupe(arena, trimmed);
+        if (!isPlausibleSymbol(sym)) return .{ .malformed = sym };
+        if (seen.contains(sym)) continue;
+        try seen.put(sym, {});
+        try out.append(arena, sym);
+    }
+    if (out.items.len == 0) return .empty;
+    if (out.items.len > max_refresh_symbols) return .{ .too_many = out.items.len };
+    return .{ .ok = out.items };
+}
+
+/// Force-refresh candles for the given symbols, synchronously.
+///
+/// The acting counterpart to `/:symbol/diagnostics`: diagnostics says a symbol is
+/// behind and nothing routine will move it, and this is the lever that moves it.
+/// Deliberately a POST, and deliberately NOT a `?refresh=1` parameter on the data
+/// endpoints - the L2 contract is that a present file is served as-is and a read
+/// never triggers a refetch, with cron as the freshness authority. Triggering the
+/// refresh path is consistent with that; making reads authoritative is not.
+///
+/// Reports whether each symbol's newest bar actually MOVED, which is the question
+/// the caller has. A successful fetch that changes nothing is the common outcome
+/// when the provider simply has no newer data, and calling that "refreshed" would
+/// hide the finding.
+pub fn handleRefresh(app: *App, req: *httpz.Request, res: *httpz.Response) !void {
+    const arena = res.arena;
+    const q = try req.query();
+    const raw = q.get("symbols") orelse {
+        res.status = 400;
+        res.body = "Missing symbols (use ?symbols=AAPL,MSFT)";
+        return;
+    };
+
+    const symbols = switch (try parseSymbolList(arena, raw)) {
+        .ok => |list| list,
+        .empty => {
+            res.status = 400;
+            res.body = "No symbols given";
+            return;
+        },
+        .malformed => |bad| {
+            res.status = 400;
+            res.body = try std.fmt.allocPrint(arena, "Invalid symbol: {s}", .{bad});
+            return;
+        },
+        .too_many => |n| {
+            res.status = 400;
+            res.body = try std.fmt.allocPrint(
+                arena,
+                "Too many symbols: {d} (limit {d})",
+                .{ n, max_refresh_symbols },
+            );
+            return;
+        },
+    };
+
+    // Non-blocking on purpose - see `App.refresh_mutex`.
+    if (!app.refresh_mutex.tryLock()) {
+        res.status = 409;
+        res.body = "A refresh is already running; retry when it finishes";
+        return;
+    }
+    defer app.refresh_mutex.unlock(app.io);
+
+    var store = zfin.cache.Store.init(app.io, arena, app.config.cache_dir);
+
+    var aw: std.Io.Writer.Allocating = .init(arena);
+    try aw.writer.writeAll("{\"results\":[");
+    for (symbols, 0..) |sym, i| {
+        if (i > 0) try aw.writer.writeByte(',');
+
+        // Read before and after rather than trusting the fetch's return value:
+        // what matters is what a client will now be served, which is whatever
+        // landed in the cache.
+        const before: ?zfin.Date = if (store.readCandleMeta(sym)) |m| m.meta.last_date else null;
+
+        if (app.svc.getCandles(sym, .{ .force_refresh = true })) |result| {
+            result.deinit();
+            const after: ?zfin.Date = if (store.readCandleMeta(sym)) |m| m.meta.last_date else null;
+            const moved = if (after) |a| (if (before) |b| b.lessThan(a) else true) else false;
+            try aw.writer.print("{{\"symbol\":\"{s}\",\"ok\":true,\"moved\":{}", .{ sym, moved });
+            if (after) |a| {
+                try aw.writer.print(",\"last_date\":\"{f}\"}}", .{a});
+            } else {
+                try aw.writer.writeAll(",\"last_date\":null}");
+            }
+        } else |err| {
+            // The error NAME, not a generic flag: an auth failure, a rate limit
+            // and a delisted ticker send the operator to different places.
+            try aw.writer.print(
+                "{{\"symbol\":\"{s}\",\"ok\":false,\"error\":\"{s}\"}}",
+                .{ sym, @errorName(err) },
+            );
+            log.warn("POST /refresh: {s} failed: {t}", .{ sym, err });
+        }
+    }
+    try aw.writer.writeAll("]}");
+
+    res.content_type = httpz.ContentType.JSON;
+    res.body = aw.written();
+}
+
 pub fn handleSrfFile(app: *App, req: *httpz.Request, res: *httpz.Response, filename: []const u8, kind: SrfKind) !void {
     return handleSrfFileByKey(app, req, res, "symbol", filename, kind);
 }
@@ -938,3 +1091,87 @@ test "groupPeerDate: an inconclusive group has no peer date" {
     // A kind with no group at all is not an error, just unanswerable.
     try std.testing.expectEqual(@as(?zfin.Date, null), groupPeerDate(report, .mutual_fund));
 }
+
+test "parseSymbolList: upper-cases, dedupes, tolerates spacing and trailing commas" {
+    var arena = std.heap.ArenaAllocator.init(std.testing.allocator);
+    defer arena.deinit();
+    const a = arena.allocator();
+
+    // The realistic invocation, including the trailing comma a shell loop leaves
+    // behind and the space a human types after one.
+    const r = try parseSymbolList(a, "nke, AMZN ,nke,");
+    const list = r.ok;
+    try std.testing.expectEqual(@as(usize, 2), list.len);
+    try std.testing.expectEqualStrings("NKE", list[0]);
+    try std.testing.expectEqualStrings("AMZN", list[1]);
+}
+
+test "parseSymbolList: nothing usable is `empty`, not a zero-length success" {
+    var arena = std.heap.ArenaAllocator.init(std.testing.allocator);
+    defer arena.deinit();
+    const a = arena.allocator();
+    // Punctuation only. Refreshing nothing and reporting success would tell the
+    // operator their symbols were handled.
+    try std.testing.expect((try parseSymbolList(a, ",,,")) == .empty);
+    try std.testing.expect((try parseSymbolList(a, "   ")) == .empty);
+    try std.testing.expect((try parseSymbolList(a, "")) == .empty);
+}
+
+test "parseSymbolList: a malformed token rejects the whole request, and is named" {
+    var arena = std.heap.ArenaAllocator.init(std.testing.allocator);
+    defer arena.deinit();
+    const a = arena.allocator();
+
+    // A path-shaped segment is the case worth blocking - these symbols reach the
+    // cache as directory names.
+    switch (try parseSymbolList(a, "AMZN,../../etc/passwd")) {
+        .malformed => |bad| try std.testing.expectEqualStrings("../../ETC/PASSWD", bad),
+        else => return error.TestUnexpectedResult,
+    }
+    // Rejecting the WHOLE list matters: refreshing AMZN and quietly dropping the
+    // typo would report success for a run that never touched what was asked for.
+    try std.testing.expect((try parseSymbolList(a, "AMZN,NK E")) == .malformed);
+    try std.testing.expect((try parseSymbolList(a, "THISTICKERISWAYTOOLONG")) == .malformed);
+}
+
+test "parseSymbolList: the cap counts DISTINCT symbols" {
+    var arena = std.heap.ArenaAllocator.init(std.testing.allocator);
+    defer arena.deinit();
+    const a = arena.allocator();
+
+    // Exactly at the limit is allowed.
+    var at: std.ArrayList(u8) = .empty;
+    defer at.deinit(a);
+    for (0..max_refresh_symbols) |i| {
+        if (i > 0) try at.append(a, ',');
+        try at.appendSlice(a, try std.fmt.allocPrint(a, "SY{d}", .{i}));
+    }
+    try std.testing.expect((try parseSymbolList(a, at.items)) == .ok);
+
+    // One past it is refused, and the count reported is the distinct count.
+    try at.appendSlice(a, ",EXTRA");
+    switch (try parseSymbolList(a, at.items)) {
+        .too_many => |n| try std.testing.expectEqual(max_refresh_symbols + 1, n),
+        else => return error.TestUnexpectedResult,
+    }
+
+    // Duplicates must NOT push a legitimate list over the cap - dedupe happens
+    // first, so this is 1 distinct symbol however many times it appears.
+    var dupes: std.ArrayList(u8) = .empty;
+    defer dupes.deinit(a);
+    for (0..max_refresh_symbols + 10) |i| {
+        if (i > 0) try dupes.append(a, ',');
+        try dupes.appendSlice(a, "AMZN");
+    }
+    const r = try parseSymbolList(a, dupes.items);
+    try std.testing.expectEqual(@as(usize, 1), r.ok.len);
+}
+
+test "refresh_mutex: a second acquire fails rather than queueing" {
+    // The 409 path. A blocking lock would serialize retries into repeated full
+    // refreshes; `tryLock` is what makes the second caller get an answer instead
+    // of a turn in the queue.
+    var m: std.Io.Mutex = .init;
+    try std.testing.expect(m.tryLock());
+    try std.testing.expect(!m.tryLock());
+}

src/main.zig

@@ -59,6 +59,12 @@ pub fn main(init: std.process.Init) !u8 {
         router.get("/", handlers.handleIndex, .{});
         router.get("/help", handlers.handleHelp, .{});
         router.get("/symbols", handlers.handleSymbols, .{});
+        // POST, not GET: this spends provider quota and writes the cache, so it is
+        // not `safe` in the HTTP sense and must not sit on a verb that proxies,
+        // prefetchers and link checkers feel free to replay. (`/:symbol/watch`
+        // stays GET only because LibreOffice's WEBSERVICE cannot issue anything
+        // else - a constraint, not a precedent.)
+        router.post("/refresh", handlers.handleRefresh, .{});
 
         // Symbol routes
         router.get("/:symbol/returns", handlers.handleReturns, .{});