%% next/kernel/host_kernel.erl — the persistent DURABLE-EXECUTION kernel the host talks to (RA-live).
%% A long-lived next/ service: boots flow_store + registers named behavior flows, then blocks in
%% http:listen so the er-scheduler (and flow_store's gen_server) stay alive across requests. The host
%% drives durable flows over HTTP; instances persist between requests (spike-proven).
%%
%% Routes (path-parameterised — binary =:= is buggy in this port, so paths are matched by byte
%% prefix, not equality):
%;;   GET /flow/start/<category>  -> start blog_digest with that category -> "<InstanceId>:<status>"
%%                                  (status = suspended | done)
%%   GET /flow/resume/<id>       -> resume instance <id> (morning timer) -> "resume:<status>"
%% This is the substrate for RA-live; TA-live adds inbox/outbox routes on the same kernel.
-module(host_kernel).
-export([start/1]).

start(Port) ->
    flow_store:start_link(),
    FF = fun (_) -> [f1, f2, f3] end,
    flow_store:register_flow(blog_digest, blog_publish_digest:build([{fetch_followers, FF}])),
    http:listen(Port, fun (Req) -> route(Req) end).

route(Req) ->
    [{status, 200}, {headers, []}, {body, respond(field(path, Req))}].

respond(P) ->
    case starts_with(P, [47,102,108,111,119,47,115,116,97,114,116,47]) of        % "/flow/start/"
        true -> do_start(last_seg(P));
        false ->
            case starts_with(P, [47,102,108,111,119,47,114,101,115,117,109,101,47]) of  % "/flow/resume/"
                true  -> do_resume(last_seg(P));
                false -> <<"path:unknown">>
            end
    end.

%% start blog_digest with the path's <category> -> "<Id>:suspended" | "<Id>:done"
do_start(CatChars) ->
    Cat = list_to_atom(CatChars),
    Env = [{activity, [{type, create}, {actor, site}, {id, <<110,49>>},
                       {object, [{type, article}, {category, Cat}]}]},
           {actor, site}],
    case flow_store:start(blog_digest, Env) of
        {ok, Id, {flow_suspended, _}} -> id_status(Id, [115,117,115,112,101,110,100,101,100]);  % "suspended"
        {ok, Id, {flow_done, _}}      -> id_status(Id, [100,111,110,101]);                       % "done"
        _                             -> <<"start:other">>
    end.

%% resume instance <id> -> "resume:done" | "resume:other"
do_resume(IdChars) ->
    case flow_store:resume(list_to_integer(IdChars), morning_ts) of
        {ok, {flow_done, _}} -> <<"resume:done">>;
        {flow_done, _}       -> <<"resume:done">>;
        _                    -> <<"resume:other">>
    end.

%% "<Id>:<statusChars>" as a response binary
id_status(Id, StatusChars) ->
    list_to_binary(integer_to_list(Id) ++ [58] ++ StatusChars).   % ++ ":" ++

%% ── plumbing (byte-level; string-literal charlists avoided for portability) ──
starts_with(Bin, Prefix) -> starts_with_(binary_to_list(Bin), Prefix).
starts_with_(_, [])            -> true;
starts_with_([C | T], [C | P]) -> starts_with_(T, P);
starts_with_(_, _)             -> false.

last_seg(Bin) -> last_seg_(binary_to_list(Bin), []).
last_seg_([], Acc)      -> lists:reverse(Acc);
last_seg_([47 | T], _)  -> last_seg_(T, []);      % reset at each '/'
last_seg_([C | T], Acc) -> last_seg_(T, [C | Acc]).

field(K, [{K, V} | _])  -> V;
field(K, [_ | Rest])    -> field(K, Rest);
field(_, [])            -> nil.