Green source tour

(ns green.advice
  "Emacs-style advice for step functions: the full set of nadvice
  combinators, LIFO stacking (most recently added outermost) with an
  optional :depth override, removal by explicit id.
  Registries are plain maps of step-name -> vector of
  {:id :how :fn :seq :depth}, so all operations are pure — advice is
  workflow-scoped, not global. A workflow embedded via
  green.workflow/step inherits its ancestors' advice at run time by
  merging registries (see `merge-entries`); the values stay pure.
  The :seq field is a caller-assigned monotonic add-order number; it lets
  a step-scoped registry and a separate all-steps registry be merged and
  sorted into one strict add-order stack (see green.workflow/run-step).
  :depth (-100..100, default 0) overrides add order the way Emacs hook
  depths do: lower pushes the advice outward, higher pushes it inward;
  at equal depth the most recently added is outermost.")

(def hows
  "The supported combinators, matching Emacs nadvice. With FUNCTION the
  advice, OLDFUN the prior chain, and r the opts map:

    :around        (FUNCTION OLDFUN r)
    :before        (FUNCTION r) then (OLDFUN r)
    :after         (OLDFUN r) then (FUNCTION r), returns OLDFUN's value
    :override      (FUNCTION r), OLDFUN never runs
    :before-while  (and (FUNCTION r) (OLDFUN r))
    :before-until  (or  (FUNCTION r) (OLDFUN r))
    :after-while   (let [ret (OLDFUN r)] (if (green-true? ret) (FUNCTION r) ret))
    :after-until   (let [ret (OLDFUN r)] (if (green-true? ret) ret (FUNCTION r)))
    :filter-args   (OLDFUN (FUNCTION r))
    :filter-return (FUNCTION (OLDFUN r))

  Typical use cases for each `how`:
    :around        dry-run, retry, timing/tracing, locks; call OLDFUN
                   zero, one, or many times
    :override      stubs, tests, or environment-specific replacements
    :before        setup/prerequisites such as backend files, dirs, locks,
                   or validation
    :after         audit, metrics, notifications, or cleanup without
                   changing OLDFUN's result
    :before-while  precondition gates: continue only while guards pass
    :before-until  fast paths/no-ops: cached or already-converged result
                   skips OLDFUN
    :after-while   success-only follow-ups such as verification,
                   registration, or notifications
    :after-until   failure recovery/fallback: rollback, repair, or supply
                   an alternate result
    :filter-args   normalize/scope inputs: defaults, derived paths,
                   sub-maps, redaction
    :filter-return normalize/enrich outputs: derived data, temp-key
                   cleanup, redaction, result-shape adaptation

  For after-while/after-until, a Green step result is true only when its
  :green/exit is 0 (missing means 0); a positive :green/exit is false.
  Non-map returns keep ordinary Clojure truthiness for compose-level use."
  #{:around :before :after :override
    :before-while :before-until :after-while :after-until
    :filter-args :filter-return})

(defn- entry [how id f order props]
  (when-not (contains? hows how)
    (throw (ex-info (str "unsupported advice combinator: " how)
                    {:how how :id id :supported hows})))
  (when-not (<= -100 (:depth props 0) 100)
    (throw (ex-info "advice :depth must be in -100..100"
                    {:depth (:depth props) :id id})))
  {:id id :how how :fn f :seq order :depth (:depth props 0)})

(defn- remove-entry-ids [entries ids]
  (filterv #(not (contains? ids (:id %))) (or entries [])))

(defn- remove-entry-id [entries id]
  (remove-entry-ids entries #{id}))

(defn add
  "Register advice `f` on `step` with combinator `how` under `id`, tagged
  with add-order `seq`. `props` may carry :depth. Re-adding an existing id
  replaces it and moves it to the top of the stack for its depth (given a
  fresh, larger `seq`)."
  [registry step how id f order props]
  (update registry step
          (fnil (fn [entries]
                  (conj (remove-entry-id entries id)
                        (entry how id f order props)))
                [])))

(defn remove-id
  "Remove the advice registered under `id` on `step`."
  [registry step id]
  (update registry step remove-entry-id id))

(defn add-global
  "Like `add`, but for a flat (not step-keyed) vector of entries — used for
  advice that applies to every step."
  [entries how id f order props]
  (conj (remove-entry-id entries id) (entry how id f order props)))

(defn remove-global-id
  "Remove the global advice registered under `id`."
  [entries id]
  (remove-entry-id entries id))

(defn merge-entries
  "Stack `outer` entries (inherited from an enclosing workflow) outside
  `inner` ones: outer seqs are rebased by `offset` so they sort above
  every inner seq, and an outer entry replaces an inner one with the same
  :id. :depth still overrides placement at compose time."
  [inner outer offset]
  (let [outer (mapv #(update % :seq + offset) outer)
        replaced-ids (set (map :id outer))]
    (into (remove-entry-ids inner replaced-ids) outer)))

(defn merge-registry
  "Merge an outer step-keyed registry over an inner one, step by step,
  with `merge-entries`."
  [inner outer offset]
  (reduce-kv (fn [reg step entries]
               (update reg step merge-entries entries offset))
             (or inner {}) outer))

(defn- green-true?
  "Truth predicate for Green step returns: success is true, failure is false.
  Compose can also be used directly with non-map values, where ordinary
  Clojure truthiness is preserved."
  [ret]
  (if (map? ret)
    (zero? (or (:green/exit ret) 0))
    (boolean ret)))

(defn- wrap [how advice-fn base]
  (case how
    :around        (fn [opts] (advice-fn base opts))
    :override      (fn [opts] (advice-fn opts))
    :before        (fn [opts] (advice-fn opts) (base opts))
    :after         (fn [opts] (let [ret (base opts)] (advice-fn opts) ret))
    :before-while  (fn [opts] (and (advice-fn opts) (base opts)))
    :before-until  (fn [opts] (or (advice-fn opts) (base opts)))
    :after-while   (fn [opts]
                     (let [ret (base opts)]
                       (if (green-true? ret) (advice-fn opts) ret)))
    :after-until   (fn [opts]
                     (let [ret (base opts)]
                       (if (green-true? ret) ret (advice-fn opts))))
    :filter-args   (fn [opts] (base (advice-fn opts)))
    :filter-return (fn [opts] (advice-fn (base opts)))))

(defn ordered
  "Entries sorted innermost-first — the order `compose` wraps them:
  higher :depth is more inward; at equal depth the most recently added
  (largest :seq) is outermost."
  [entries]
  (sort-by (juxt #(- (:depth % 0)) :seq) entries))

(defn compose
  "Wrap `base` with `entries`, ordered like Emacs nadvice: lower :depth is
  more outward; at equal depth the most recently added (largest :seq) is
  outermost. The reduce builds inside-out over `ordered` (innermost-first)
  entries."
  [base entries]
  (reduce (fn [g {:keys [how] advice-fn :fn}]
            (wrap how advice-fn g))
          base
          (ordered entries)))

(ns green.ansible
  "Event-aware Ansible steps: any non-:delete event (conventionally :create)
  runs the create playbook, :delete runs the delete playbook — both via
  `ansible-playbook` over SSH. After a successful run the PLAY RECAP is
  parsed and merged into opts under a namespaced key (:ansible/recap by
  default). The inventory is not hardwired: attach `inventory-advice` as a
  :before advice to write an INI inventory from a function of opts before
  the step runs, the way green.tofu attaches backends."
  (:require [cheshire.core :as json]
            [clojure.java.io :as io]
            [clojure.java.shell :as sh]
            [clojure.string :as str]))

(def default-playbooks
  "Event -> playbook file, relative to the step's :dir."
  {:create "create.yml" :delete "delete.yml"})

(defn playbook
  "The playbook `ansible-step` runs for opts' :green/event: :delete selects
  the :delete entry, every other event the :create entry."
  ([opts] (playbook opts default-playbooks))
  ([opts playbooks]
   (let [playbooks (merge default-playbooks playbooks)]
     (if (= :delete (:green/event opts))
       (:delete playbooks)
       (:create playbooks)))))

(def ^:private recap-line-re
  #"(?m)^(\S+)\s+:\s+ok=(\d+)\s+changed=(\d+)\s+unreachable=(\d+)\s+failed=(\d+)\s+skipped=(\d+)\s+rescued=(\d+)\s+ignored=(\d+)")

(defn parse-recap
  "Parse the PLAY RECAP section of `ansible-playbook` output into
  {host {:ok n :changed n :unreachable n :failed n :skipped n :rescued n
  :ignored n}}."
  [out]
  (into {}
        (map (fn [[_ host & counts]]
               [host (zipmap [:ok :changed :unreachable :failed
                              :skipped :rescued :ignored]
                             (map parse-long counts))]))
        (re-seq recap-line-re (str out))))

(defn- env-with [extra]
  (merge (into {} (System/getenv)) extra))

(defn- ansible! [dir env & args]
  (apply sh/sh "ansible-playbook" (concat args [:dir dir :env env])))

(defn- fail [opts {:keys [exit err out]} pb]
  (assoc opts
         :green/exit exit
         :green/err (str "ansible-playbook " pb " failed: "
                         (or (not-empty out) (not-empty err) "(no output)"))))

(defn ansible-step
  "Run `ansible-playbook` in `dir` according to :green/event (see
  `playbook`). On success the parsed PLAY RECAP is merged under `recap-key`
  (default :ansible/recap) — never top-level.

  Options:
    :dir               working directory; playbook and inventory paths are
                       resolved relative to it
    :inventory         inventory file (default \"inventory.ini\")
    :playbooks         {:create ... :delete ...} overriding `default-playbooks`
    :private-key       SSH private key file passed as --private-key
    :user              remote user passed as -u
    :extra-vars        map passed as -e in JSON form
    :host-key-checking set to false to export ANSIBLE_HOST_KEY_CHECKING=False
                       for the run — for ephemeral or emulated hosts whose
                       host keys change on every create. Omitted or true,
                       the environment is left untouched.
    :recap-key         namespaced key for the parsed recap"
  [opts {:keys [dir inventory playbooks private-key user extra-vars
                host-key-checking recap-key]
         :or {inventory "inventory.ini" recap-key :ansible/recap}}]
  (let [pb (playbook opts playbooks)
        env (cond-> nil
              (false? host-key-checking)
              (assoc "ANSIBLE_HOST_KEY_CHECKING" "False"))
        args (cond-> ["-i" inventory]
               private-key (into ["--private-key" (str private-key)])
               user (into ["-u" user])
               extra-vars (into ["-e" (json/generate-string extra-vars)])
               true (conj pb))
        res (apply ansible! dir (some-> env env-with) args)]
    (if (pos? (:exit res))
      (fail opts res pb)
      (assoc opts :green/exit 0 recap-key (parse-recap (:out res))))))

;; --- inventory ------------------------------------------------------------

(defn- ini-name [x]
  (if (keyword? x) (name x) (str x)))

(defn- ini-vars [vars]
  (map (fn [[k v]] (str (ini-name k) "=" (ini-name v)))
       (sort-by (comp ini-name key) vars)))

(defn- host-line [{host :name vars :vars}]
  (str/join " " (cons (ini-name host) (ini-vars vars))))

(defn- group-section [[group {:keys [hosts vars]}]]
  (cond-> (str "[" (ini-name group) "]\n"
               (str/join "" (map #(str (host-line %) "\n") hosts)))
    (seq vars) (str "\n[" (ini-name group) ":vars]\n"
                    (str/join "" (map #(str % "\n") (ini-vars vars))))))

(defn inventory-ini
  "Render an Ansible INI inventory from
  {group {:hosts [{:name \"zk1\" :vars {:ansible_host \"172.17.0.3\"}} ...]
          :vars {:ansible_user \"root\"}}}.
  Groups and vars are emitted in sorted order for deterministic output."
  [groups]
  (str/join "\n" (map group-section (sort-by (comp ini-name key) groups))))

(defn- resolve-config [config opts]
  (if (fn? config) (config opts) config))

(defn inventory-advice
  "Build a :before advice that writes an INI inventory to the file returned
  by (file-fn opts) before the step runs. `groups` is the inventory data
  (see `inventory-ini`), or a function of opts returning it."
  [file-fn groups]
  (fn [opts]
    (let [f (io/file (file-fn opts))]
      (io/make-parents f)
      (spit f (inventory-ini (resolve-config groups opts))))
    opts))

(ns green.cli
  "CLI plumbing: `./green <event> [-f|--file green.edn] [--start step]
  [--end step]`. The first positional argument is the lifecycle event,
  stamped into opts as :green/event. --start/--end run a slice of the graph."
  (:require [babashka.cli :as cli]
            [clojure.edn :as edn]
            [clojure.java.io :as io]
            [green.workflow :as wf]))

(def ^:private cli-spec
  {:file {:alias :f :default "green.edn" :desc "Desired state EDN file"}
   :start {:coerce :keyword :desc "Override the workflow start step"}
   :end {:coerce :keyword :desc "Override the workflow end step (slice boundary)"}
   :dry-run {:coerce :boolean :desc "Stamp :green/dry-run — steps advised with green.dry-run are skipped"}})

(def usage
  "Usage: green <event> [-f|--file green.edn] [--start step] [--end step] [--dry-run]")

(defn run-cli
  "Parse `args`, load the desired state, stamp :green/event, run `workflow`.
  Returns the final opts map (:green/exit 2 on usage/state-file errors)."
  ([workflow] (run-cli workflow *command-line-args*))
  ([workflow args]
   (try
     (let [{:keys [args opts]} (cli/parse-args (vec args) {:spec cli-spec})
           event (first args)]
       (if-not event
         {:green/exit 2 :green/err usage}
         (let [file (io/file (:file opts))]
           (if-not (.exists file)
             {:green/exit 2 :green/err (str "desired state file not found: " file)}
             (let [state (edn/read-string (slurp file))
                   workflow (cond-> workflow
                              (:start opts) (assoc :green.workflow/start (:start opts))
                              (:end opts) (assoc :green.workflow/end (:end opts)))]
               (wf/run workflow
                       (cond-> (assoc state :green/event (keyword event))
                         (:dry-run opts) (assoc :green/dry-run true))))))))
     (catch Throwable t
       {:green/exit 2 :green/err (or (ex-message t) (str (class t)))}))))

(defn exec
  "Run and exit the process with :green/exit, printing :green/err and
  :green/trace to stderr. For use from the project's babashka script."
  ([workflow] (exec workflow *command-line-args*))
  ([workflow args]
   (let [{:green/keys [exit err trace]} (run-cli workflow args)]
     (when err
       (binding [*out* *err*]
         (println err)
         (when trace (println trace))))
     (System/exit (or exit 0)))))

(ns green.dry-run
  "Dry-run support, built on the advice facility rather than hardwired into
  steps. Attach `advice` (an :around) to side-effecting steps — or `advise`
  a whole list of them; when :green/dry-run is set in opts (the CLI's
  --dry-run flag stamps it) the step is skipped with a note instead of run."
  (:require [green.workflow :as wf]))

(defn- logln [& xs]
  (locking *out*
    (apply println xs)
    (flush)))

(defn advice
  "Build an :around advice for `step`: when :green/dry-run is set, print
  what would have run and skip the base function; otherwise call through."
  [step]
  (fn [f opts]
    (if (:green/dry-run opts)
      (do (logln (str "dry-run: would run " step
                       " (" (name (:green/event opts :create)) ")"))
          (assoc opts :green/exit 0))
      (f opts))))

(defn advise
  "Attach dry-run advice to every step in `steps` under id ::skip.
  Returns the advised workflow; remove per step with
  (wf/advice-remove wf step :green.dry-run/skip)."
  [wf steps]
  (reduce (fn [w s] (wf/advice-add w s :around ::skip (advice s))) wf steps))

(ns green.progress
  "Progress reporting, built on the advice facility. Attach with `advise`
  to print step start/end with elapsed time; reads `:green/step` from opts."
  (:require [green.workflow :as wf]))

(defn- logln [& xs]
  (locking *out*
    (apply println xs)
    (flush)))

(defn progress
  "An :around advice that prints step name on entry and elapsed time on exit.
  Reads :green/step from opts (stamped by the engine)."
  [f opts]
  (let [step (:green/step opts)
        event (name (:green/event opts :create))]
    (logln (str ">>> " step " (" event ")"))
    (let [t0 (System/currentTimeMillis)
          result (f opts)
          ms (- (System/currentTimeMillis) t0)]
      (logln (str "<<< " step " (" ms "ms)"))
      result)))

(defn advise
  "Attach progress advice to every step. Returns the advised workflow."
  [wf]
  (wf/advice-add-all wf :around ::progress progress))

(ns green.scaffold
  "Flat file-spec scaffolding DSL. A spec is a seq of maps, one per file:

    {:template :zk/main.tf                    ; classpath resource zk/main.tf
     :target   \"{{workdir}}/n/{{node.id}}/main.tf\" ; Selmer-rendered vs :data
     :data     {...}}

  On :green/event :delete the same specs name the targets to remove."
  (:require [clojure.java.io :as io]
            [clojure.string :as str]
            [selmer.parser :as selmer]))

(defn template-path
  "Resource path for a qualified template keyword: namespace dots become
  directories, the name is used verbatim. :my.app/zoo.cfg -> my/app/zoo.cfg"
  [kw]
  (str (some-> (namespace kw) (str/replace "." "/") (str "/"))
       (name kw)))

(defn render-template
  "Render the classpath template named by qualified keyword `kw` with `data`."
  [kw data]
  (let [path (template-path kw)
        res (io/resource path)]
    (when-not res
      (throw (ex-info (str "template not found on classpath: " path)
                      {:template kw :path path})))
    (selmer/render (slurp res) data)))

(defn- prune-empty-dir! [^java.io.File f]
  (let [p (.getParentFile f)]
    (when (and p (.isDirectory p) (empty? (.list p)))
      (.delete p))))

(defn- target-path [{:keys [target data]}]
  (selmer/render target data))

(defn- target-paths [specs]
  (mapv target-path specs))

(defn- delete-target! [target]
  (let [f (io/file target)]
    (when (.exists f) (io/delete-file f))
    (prune-empty-dir! f)))

(defn- create-target! [{:keys [template data]} target]
  (let [f (io/file target)]
    (io/make-parents f)
    (spit f (render-template template data))))

(defn- delete-targets! [targets]
  (doseq [target targets]
    (delete-target! target)))

(defn- create-targets! [specs targets]
  (doseq [[spec target] (map vector specs targets)]
    (create-target! spec target)))

(defn scaffold
  "Materialize `specs` (create) or remove their targets (delete), driven by
  :green/event in `opts`. Returns opts with :green/exit 0 and the affected
  paths under :green.scaffold/written or :green.scaffold/deleted."
  [opts specs]
  (let [specs (vec specs)
        targets (target-paths specs)]
    (if (= :delete (:green/event opts))
      (do (delete-targets! targets)
          (assoc opts :green/exit 0 :green.scaffold/deleted targets))
      (do (create-targets! specs targets)
          (assoc opts :green/exit 0 :green.scaffold/written targets)))))

(ns green.tofu
  "Event-aware OpenTofu steps: any non-:delete event (conventionally
  :create) -> init + apply, :delete -> init + destroy. After apply,
  `tofu output -json` is merged into opts under a namespaced key
  (:tofu/outputs by default). The backend is not hardwired: attach a
  :before advice built by `backend-advice`, `local-backend-advice`,
  `s3-backend-advice`, or `gcs-backend-advice` to write backend.tf before
  the tofu command runs."
  (:require [cheshire.core :as json]
            [clojure.java.io :as io]
            [clojure.java.shell :as sh]))

(def ^:private init-args ["init" "-input=false" "-no-color"])

(def ^:private apply-args ["apply" "-auto-approve" "-input=false" "-no-color"])

(def ^:private destroy-args ["destroy" "-auto-approve" "-input=false" "-no-color"])

(defn- tofu! [dir & args]
  (apply sh/sh "tofu" (concat args [:dir dir])))

(defn- action-args [delete?]
  (if delete? destroy-args apply-args))

(defn- failed? [{:keys [exit]}]
  (pos? exit))

(defn- fail [opts {:keys [exit err out]} cmd]
  (assoc opts
         :green/exit exit
         :green/err (str "tofu " cmd " failed: "
                         (or (not-empty err) (not-empty out) "(no output)"))))

(defn- parse-outputs [out]
  (into {}
        (map (fn [[k v]] [(keyword k) (get v "value")]))
        (json/parse-string out)))

(defn outputs
  "Parse `tofu output -json` in `dir` into a plain map of keyword -> value."
  [dir]
  (let [{:keys [exit out err]} (tofu! dir "output" "-json")]
    (when (pos? exit)
      (throw (ex-info (str "tofu output failed: " err) {:dir dir})))
    (parse-outputs out)))

(defn tofu-step
  "Run OpenTofu in `dir` according to :green/event. On success, apply merges
  the outputs under `output-key` (default :tofu/outputs) — never top-level."
  [opts {:keys [dir output-key] :or {output-key :tofu/outputs}}]
  (let [delete? (= :delete (:green/event opts))
        init (apply tofu! dir init-args)]
    (if (failed? init)
      (fail opts init "init")
      (let [cmd (action-args delete?)
            res (apply tofu! dir cmd)]
        (cond
          (failed? res) (fail opts res (first cmd))
          delete? (assoc opts :green/exit 0)
          :else (assoc opts :green/exit 0 output-key (outputs dir)))))))

(defn- hcl-value [v]
  (cond
    (keyword? v) (pr-str (name v))
    (boolean? v) (str v)
    (number? v) (str v)
    :else (pr-str (str v))))

(defn- hcl-attribute [[k v]]
  (str "    " (name k) " = " (hcl-value v) "\n"))

(defn- backend-hcl [type config]
  (str "terraform {\n  backend \"" type "\" {\n"
       (apply str (map hcl-attribute (sort-by key config)))
       "  }\n}\n"))

(defn- resolve-config [config opts]
  (if (fn? config) (config opts) config))

(defn- write-backend! [dir type config]
  (.mkdirs dir)
  (spit (io/file dir "backend.tf") (backend-hcl type config)))

(defn backend-advice
  "Build a :before advice that writes a backend config into the directory
  returned by (dir-fn opts) before the step runs. `type` is the backend
  name (\"local\", \"s3\", \"gcs\", …); `config` is a flat map of backend
  attributes, or a function of opts returning one."
  [dir-fn type config]
  (fn [opts]
    (write-backend! (io/file (dir-fn opts)) type (resolve-config config opts))
    opts))

(defn local-backend-advice
  "Backend advice for the local filesystem backend."
  ([dir-fn] (local-backend-advice dir-fn {}))
  ([dir-fn config] (backend-advice dir-fn "local" config)))

(defn s3-backend-advice
  "Backend advice for the S3 backend, e.g.
  {:bucket \"my-state\" :key \"green/node-1.tfstate\" :region \"eu-west-1\"}."
  [dir-fn config]
  (backend-advice dir-fn "s3" config))

(defn gcs-backend-advice
  "Backend advice for the GCS backend, e.g.
  {:bucket \"my-state\" :prefix \"green/node-1\"}."
  [dir-fn config]
  (backend-advice dir-fn "gcs" config))

(ns green.workflow
  "The workflow engine: a graph of steps threaded by an opts map.

  - wire-fn: (step run-opts) -> [fn & next-steps] — the static
    happy-path graph for this run. It may depend on stable run-level inputs
    such as :green/event. Multiple successors run in parallel.
  - next-fn (optional): (next-fn step default-next opts) -> seq of
    [next-step opts] pairs — dynamic routing, error branching, fan-out.
  - Steps report outcome via :green/exit (0 ok, >0 error), :green/err,
    :green/trace. Thrown exceptions are caught and converted.
  - Branches converging on the same step join: the join step runs once
    with the fork-point opts plus :green/branches (vector of branch results).
  - A branch failing inside a fork lets the in-flight siblings finish their
    current step, then the fork collapses: the join is skipped and the worst
    exit propagates, with all branch results under :green/branches.
  - advice-add attaches advice to one step; advice-add-all attaches advice
    to every step. Both stack in strict add order (most recently added,
    from either, is outermost), unless a :depth prop overrides placement
    (lower = more outward), as in Emacs.
  - Advice is inherited across `step` embeds: a run stamps its effective
    advice into opts and a nested run merges it over the child's own —
    step names match flat at any depth, ancestor advice is outermost, and
    an ancestor entry replaces a same-id child entry. advice-plan shows
    the composed stack for a step."
  (:require [green.advice :as advice])
  (:import [java.io PrintWriter StringWriter]))

(defn workflow
  "Construct a workflow. :start is required; :end is an optional slice
  boundary (it runs, then the workflow stops); :wire-fn is required and
  is called as (wire-fn step run-opts); :next-fn is optional."
  [{:keys [start end wire-fn next-fn]}]
  (when-not (keyword? start)
    (throw (ex-info "workflow :start must be a keyword" {:start start})))
  (when-not (fn? wire-fn)
    (throw (ex-info "workflow :wire-fn must be a function" {:wire-fn wire-fn})))
  {::start start ::end end ::wire-fn wire-fn ::next-fn next-fn
   ::advice {} ::advice-all [] ::advice-seq 0})

(defn- next-seq [wf]
  (::advice-seq wf 0))

(defn advice-add
  "Return a workflow with advice `f` added on `step` (combinator `how`,
  explicit `id`). Pure — the original workflow is untouched. Composes with
  any all-steps advice (see `advice-add-all`) in strict add order: whichever
  was added more recently is outermost. `props` may carry :depth
  (-100..100, default 0), which overrides add order: lower depth pushes the
  advice outward, higher pushes it inward, as with Emacs hook depths."
  ([wf step how id f] (advice-add wf step how id f nil))
  ([wf step how id f props]
   (let [s (next-seq wf)]
     (-> wf
         (update ::advice advice/add step how id f s props)
         (assoc ::advice-seq (inc s))))))

(defn advice-remove
  "Return a workflow with the advice registered under `id` on `step` removed."
  [wf step id]
  (update wf ::advice advice/remove-id step id))

(defn advice-add-all
  "Return a workflow with advice `f` added on every step (combinator `how`,
  explicit `id`). Pure — the original workflow is untouched. Composes with
  any per-step advice in strict add order: whichever was added more
  recently is outermost, regardless of whether it came from `advice-add`
  or `advice-add-all`. `props` may carry :depth, as in `advice-add`."
  ([wf how id f] (advice-add-all wf how id f nil))
  ([wf how id f props]
   (let [s (next-seq wf)]
     (-> wf
         (update ::advice-all advice/add-global how id f s props)
         (assoc ::advice-seq (inc s))))))

(defn advice-remove-all
  "Return a workflow with the all-steps advice registered under `id` removed."
  [wf id]
  (update wf ::advice-all advice/remove-global-id id))

;; --- advice inheritance across embeds -------------------------------------
;; A run stamps its effective registries into every step's opts under
;; ::inherited; `step` forwards the stamp into the nested run, whose
;; `inherit` merges it over the child's own advice. Inheritance is
;; transitive because each run stamps its already-merged registries.

(defn- inherit
  "Merge an `inherited` registry payload {:advice :advice-all} from an
  enclosing run over `wf`'s own advice: inherited entries stack outside
  the child's own, and an inherited entry replaces a same-id child entry
  (per step, or in the all-steps list). Step names are flat — whatever an
  ancestor advised under a name applies to this workflow's step of that
  name."
  [wf inherited]
  (if-let [{:keys [advice advice-all]} inherited]
    (let [n (next-seq wf)]
      (-> wf
          (update ::advice advice/merge-registry advice n)
          (update ::advice-all advice/merge-entries advice-all n)))
    wf))

(defn advice-plan
  "Debugging: the advice stack that would wrap `step` at run time,
  outermost first. `wfs` is a single workflow or a chain
  [outermost ... innermost] — the workflows a run traverses to reach
  `step` through `green.workflow/step` embeds (e.g. [parent-wf cluster-wf]
  for an embedded :zk/node). Returns
  [{:id :how :depth :seq :scope :level} ...] where :scope is :step or
  :all and :level indexes the chain element that registered the advice
  (0 = outermost). A child entry replaced by a same-id ancestor entry
  does not appear."
  [wfs step]
  (let [chain (if (map? wfs) [wfs] (vec wfs))
        tag (fn [wf level]
              (-> wf
                  (update ::advice
                          (fn [m]
                            (into {} (map (fn [[k es]]
                                            [k (mapv #(assoc % :scope :step :level level) es)]))
                              m)))
                  (update ::advice-all
                          (fn [es] (mapv #(assoc % :scope :all :level level) es)))))
        eff (reduce (fn [inherited [level wf]]
                      (let [wf (inherit (tag wf level) inherited)]
                        {:advice (::advice wf) :advice-all (::advice-all wf)}))
                    nil
                    (map-indexed vector chain))
        entries (concat (:advice-all eff) (get (:advice eff) step))]
    (->> (advice/ordered entries)
         reverse
         (mapv #(select-keys % [:id :how :depth :seq :scope :level])))))

;; --- static graph (for join scheduling) ---------------------------------

(defn- static-successors [wire-fn step run-opts]
  ;; Deliberately lenient: this graph exists only for join detection, so a
  ;; name the wire-fn can't resolve (nil or a throw) just contributes no
  ;; static edges — next-fn may own routing for it. Real wiring bugs still
  ;; surface when the step runs: run-step calls the same wire-fn and converts
  ;; the failure to :green/exit. Throwing here would escape run uncaught,
  ;; bypassing the Unix-style outcome contract.
  (try (rest (wire-fn step run-opts)) (catch Exception _ nil)))

(defn- static-graph [wire-fn start run-opts]
  (loop [g {} frontier [start]]
    (if-let [s (first frontier)]
      (if (contains? g s)
        (recur g (subvec frontier 1))
        (let [succ (vec (static-successors wire-fn s run-opts))]
          (recur (assoc g s succ) (into (subvec frontier 1) succ))))
      g)))

(defn- reaches?
  "Can a branch currently at `from` (about to run it) later arrive at `to`,
  following static wire-fn edges?"
  [g from to]
  (loop [seen #{} frontier (vec (get g from))]
    (if (empty? frontier)
      false
      (let [s (peek frontier) frontier (pop frontier)]
        (cond
          (= s to) true
          (seen s) (recur seen frontier)
          :else (recur (conj seen s) (into frontier (get g s))))))))

;; --- running one step ----------------------------------------------------

(defn- stack-trace [^Throwable t]
  (let [sw (StringWriter.)]
    (.printStackTrace t (PrintWriter. sw true))
    (str sw)))

(defn- failed? [opts]
  (pos? (:green/exit opts 0)))

(defn- step-failure [opts ^Throwable t]
  (assoc opts
         :green/exit (or (:green/exit (ex-data t)) 1)
         :green/err (or (ex-message t) (str (class t)))
         :green/trace (stack-trace t)))

(defn- scheduler-failure [opts ^Throwable t]
  (assoc opts
         :green/exit 1
         :green/err (or (ex-message t) (str (class t)))
         :green/trace (stack-trace t)))

(defn- with-default-exit [opts]
  (cond-> opts
    (nil? (:green/exit opts)) (assoc :green/exit 0)))

(defn- inherited-payload [wf]
  {:advice (::advice wf)
   :advice-all (::advice-all wf)})

(defn- stamp-inherited [wf opts]
  (assoc opts ::inherited (inherited-payload wf)))

(defn- step-advice [wf step]
  (concat (::advice-all wf) (get-in wf [::advice step])))

(defn- run-step [wf step run-opts opts]
  (try
    (let [decl ((::wire-fn wf) step run-opts)
          f (first decl)]
      (when-not (fn? f)
        (throw (ex-info (str "no function wired for step " step) {:step step})))
      (let [ret ((advice/compose f (step-advice wf step))
                 (assoc (stamp-inherited wf opts) :green/step step))]
        (when-not (map? ret)
          (throw (ex-info (str "step " step " returned a non-map: " (pr-str ret))
                          {:step step})))
        (with-default-exit (dissoc ret ::inherited))))
    (catch Throwable t
      (step-failure opts t))))

(defn- next-pairs
  "Successor [step opts] pairs for `step` after it produced `opts`.
  The end step is a hard boundary; without next-fn an error halts."
  [wf step run-opts opts]
  (if (= step (::end wf))
    []
    (let [dn (seq (rest ((::wire-fn wf) step run-opts)))]
      (if-let [nf (::next-fn wf)]
        (vec (nf step dn opts))
        (if (failed? opts)
          []
          (mapv (fn [s] [s opts]) dn))))))

;; --- the scheduler --------------------------------------------------------

;; Live entries: {:step k :opts m :parent id :forks [frame…]} where a fork
;; frame is {:id fork-id :opts fork-point-opts}. :parent identifies the
;; run-unit that produced the entry, so same-step entries from one fan-out
;; run individually while entries converging from different origins join.
;; Finished branches are {:opts m :forks [frame…]}; the frames let a failure
;; collapse its enclosing fork.

(defn- children [uid opts pairs forks]
  (cond
    (empty? pairs)
    {:terminals [{:opts opts :forks forks}]}

    (= 1 (count pairs))
    (let [[s o] (first pairs)]
      {:new-entries [{:step s :opts o :parent uid :forks forks}]})

    :else
    (let [frame {:id uid :opts opts}]
      {:new-entries (mapv (fn [[s o]]
                            {:step s :opts o :parent uid :forks (conj forks frame)})
                          pairs)})))

(defn- terminal-result [opts forks]
  {:terminals [{:opts opts :forks forks}]})

(defn- branch-worst-exit [branch-opts]
  (apply max (map #(:green/exit % 0) branch-opts)))

(defn- first-failed-branch [branch-opts]
  (first (filter failed? branch-opts)))

(defn- join-forks [entries]
  (or (some #(when (seq (:forks %)) (:forks %)) entries) []))

(defn- failed-join-result [fork-opts forks branch-opts worst]
  (let [bad (first-failed-branch branch-opts)]
    (terminal-result (assoc fork-opts
                            :green/exit worst
                            :green/err (:green/err bad)
                            :green/trace (:green/trace bad)
                            :green/branches branch-opts)
                     forks)))

(defn- run-single-unit [wf uid step run-opts {:keys [opts forks]}]
  (let [opts' (run-step wf step run-opts opts)]
    (children uid opts' (next-pairs wf step run-opts opts') forks)))

(defn- run-join-unit [wf uid step run-opts entries]
  (let [branch-opts (mapv :opts entries)
        forks (join-forks entries)
        fork-opts (if (seq forks) (:opts (peek forks)) (first branch-opts))
        forks' (if (seq forks) (pop forks) forks)
        worst (branch-worst-exit branch-opts)]
    (if (pos? worst)
      (failed-join-result fork-opts forks' branch-opts worst)
      (let [opts' (run-step wf step run-opts (assoc fork-opts :green/branches branch-opts))]
        (children uid opts' (next-pairs wf step run-opts opts') forks')))))

(defn- unit-base-opts [entry entries]
  (or (:opts entry) (:opts (first entries)) {}))

(defn- unit-forks [entry entries]
  (or (:forks entry) (:forks (first entries)) []))

(defn- failed-unit-result [entry entries ^Throwable t]
  (terminal-result (scheduler-failure (unit-base-opts entry entries) t)
                   (unit-forks entry entries)))

(defn- run-unit [wf run-opts {:keys [kind step entry entries]}]
  (let [uid (gensym "green-unit")]
    (try
      (case kind
        :single (run-single-unit wf uid step run-opts entry)
        :join (run-join-unit wf uid step run-opts entries))
      (catch Throwable t
        (failed-unit-result entry entries t)))))

(defn- failed-fork-branch? [branch]
  (and (failed? (:opts branch)) (seq (:forks branch))))

(defn- in-fork? [fork-id entry]
  (boolean (some #(= fork-id (:id %)) (:forks entry))))

(defn- fork-members [fork-id live finished]
  (into (filterv (partial in-fork? fork-id) finished)
        (filterv (partial in-fork? fork-id) live)))

(defn- outside-fork [fork-id entries]
  (filterv (complement (partial in-fork? fork-id)) entries))

(defn- collapsed-fork-entry [fork-opts bad branch-opts worst]
  (let [worst-opts (first (filter #(= worst (:green/exit % 0)) branch-opts))]
    {:opts (assoc fork-opts
                  :green/exit worst
                  :green/err (:green/err worst-opts)
                  :green/trace (:green/trace worst-opts)
                  :green/branches branch-opts)
     :forks (pop (:forks bad))}))

(defn- collapse-fork [live finished bad]
  (let [{fork-id :id fork-opts :opts} (peek (:forks bad))
        branch-opts (mapv :opts (fork-members fork-id live finished))
        worst (branch-worst-exit branch-opts)]
    [(outside-fork fork-id live)
     (conj (outside-fork fork-id finished)
           (collapsed-fork-entry fork-opts bad branch-opts worst))]))

(defn- collapse-doomed
  "While a finished branch failed inside a fork, collapse that fork: absorb
  its live entries (their current opts are their branch results — siblings
  finished their step, nothing new starts) and its finished branches, skip
  the join, and emit a terminal carrying the worst exit and :green/branches.
  Cascades outward through nested forks."
  [live finished]
  (loop [live live finished finished]
    (if-let [bad (first (filter failed-fork-branch? finished))]
      (let [[live' finished'] (collapse-fork live finished bad)]
        (recur live' finished'))
      [live finished])))

(defn- finalize [finished]
  (let [terminals (mapv #(dissoc (:opts %) ::inherited) finished)]
    (cond
      (empty? terminals) {:green/exit 0}
      (= 1 (count terminals)) (first terminals)
      :else (or (first (filter failed? terminals))
                (last terminals)))))

(defn- blocked-step? [g live step]
  (some #(and (not= (:step %) step) (reaches? g (:step %) step))
        live))

(defn- ready-steps [g live by-step]
  (or (seq (remove #(blocked-step? g live %) (keys by-step)))
      (keys by-step)))

(defn- waiting-steps [by-step ready]
  (remove (set ready) (keys by-step)))

(defn- entries-for-steps [by-step steps]
  (mapcat #(get by-step %) steps))

(defn- same-origin? [entries]
  (or (= 1 (count entries))
      (apply = (map :parent entries))))

(defn- single-units [step entries]
  (map (fn [entry] {:kind :single :step step :entry entry}) entries))

(defn- step-units [step entries]
  (if (same-origin? entries)
    (single-units step entries)
    [{:kind :join :step step :entries entries}]))

(defn- ready-units [by-step ready]
  (mapcat (fn [step] (step-units step (get by-step step))) ready))

(defn- run-units [wf run-opts units]
  (mapv deref (mapv (fn [unit] (future (run-unit wf run-opts unit))) units)))

(defn- scheduler-step [wf run-opts g live finished]
  (let [by-step (group-by :step live)
        ready (ready-steps g live by-step)
        waiting (waiting-steps by-step ready)
        units (ready-units by-step ready)
        results (run-units wf run-opts units)]
    (collapse-doomed
     (-> []
         (into (entries-for-steps by-step waiting))
         (into (mapcat :new-entries results)))
     (into finished (mapcat :terminals results)))))

(declare run)

(defn step
  "Turn a workflow into a step function (opts -> opts), so workflows compose
  into higher-level workflows: wire the result like any other step, advise
  it, fan it out. The sub-workflow's :green/exit propagates naturally, and
  ambient keys like :green/event and :green/dry-run flow in with opts.

  The enclosing run's advice is inherited: the nested run merges it over
  the sub-workflow's own (see the ns docstring). The engine re-stamps the
  inherited registry after :in runs, so :in may build sub-opts from
  scratch without severing inheritance.

  Options:
    :in  (fn [opts] sub-opts)        — shape the opts entering the sub-workflow
    :out (fn [opts sub-result] opts) — merge the sub-result back into the
                                       parent's opts (default: the sub-result
                                       itself is the step's result)"
  ([wf] (step wf {}))
  ([wf {:keys [in out]}]
   (fn [opts]
     (let [inherited (::inherited opts)
           sub-opts (cond-> ((or in identity) opts)
                      inherited (assoc ::inherited inherited))
           result (run wf sub-opts)]
       (if out (out opts result) result)))))

(defn run
  "Run the workflow from its start step with `opts` as the initial state.
  The same initial opts are passed to `wire-fn` as `run-opts` for the
  whole run, so the static graph stays stable. Returns the final opts map;
  its :green/exit is the workflow's exit code. When `opts` carries an
  inherited advice registry (stamped by an enclosing run through `step`),
  it is merged over the workflow's own advice before anything runs."
  [wf opts]
  (let [run-opts opts
        wf (inherit wf (::inherited opts))
        g (static-graph (::wire-fn wf) (::start wf) run-opts)]
    (loop [live [{:step (::start wf) :opts opts :parent ::root :forks []}]
           finished []]
      (if (empty? live)
        (finalize finished)
        (let [[live' finished'] (scheduler-step wf run-opts g live finished)]
          (recur live' finished'))))))

(ns green.advice-test
  (:require [clojure.test :refer [deftest is testing]]
            [green.advice :as advice]
            [green.workflow :as wf]))

(defn- log [o x] (update o :log (fnil conj []) x))

(defn- single-step-wf []
  (wf/workflow {:start :t/step
                :wire-fn (fn [_ _] [(fn [o] (log o :base))])}))

(deftest filter-return-and-lifo-stacking
  (let [base (single-step-wf)
        advised (-> base
                    (wf/advice-add :t/step :filter-return ::a #(log % :a))
                    (wf/advice-add :t/step :filter-return ::b #(log % :b)))]
    (testing "most recently added advice is outermost"
      (is (= [:base :a :b] (:log (wf/run advised {})))))
    (testing "the original workflow is untouched (workflow-scoped advice)"
      (is (= [:base] (:log (wf/run base {})))))))

(deftest advice-remove-by-id
  (let [advised (-> (single-step-wf)
                    (wf/advice-add :t/step :filter-return ::a #(log % :a))
                    (wf/advice-add :t/step :filter-return ::b #(log % :b))
                    (wf/advice-remove :t/step ::a))]
    (is (= [:base :b] (:log (wf/run advised {}))))))

(deftest re-adding-same-id-replaces-and-moves-to-top
  (let [advised (-> (single-step-wf)
                    (wf/advice-add :t/step :filter-return ::a #(log % :a1))
                    (wf/advice-add :t/step :filter-return ::b #(log % :b))
                    (wf/advice-add :t/step :filter-return ::a #(log % :a2)))]
    (is (= [:base :b :a2] (:log (wf/run advised {}))))))

(deftest override-replaces-base
  (let [advised (-> (single-step-wf)
                    (wf/advice-add :t/step :override ::o #(log % :override)))]
    (is (= [:override] (:log (wf/run advised {}))))))

(deftest around-controls-the-call
  (let [advised (-> (single-step-wf)
                    (wf/advice-add :t/step :around ::ar
                                   (fn [f o] (log (f (log o :in)) :out))))]
    (is (= [:in :base :out] (:log (wf/run advised {}))))))

(deftest filter-args-transforms-input
  (let [advised (-> (single-step-wf)
                    (wf/advice-add :t/step :filter-args ::fa #(assoc % :x 1)))]
    (is (= 1 (:x (wf/run advised {}))))))

(deftest before-and-after-run-for-side-effects
  (let [seen (atom [])
        advised (-> (single-step-wf)
                    (wf/advice-add :t/step :before ::b
                                   (fn [o] (swap! seen conj [:before (:log o)]) o))
                    (wf/advice-add :t/step :after ::a
                                   (fn [o] (swap! seen conj [:after (:log o)]) o)))
        res (wf/run advised {})]
    (testing "before and after both see the step's input opts (Emacs: both
              get r); their returns are ignored and OLDFUN's value flows out"
      (is (= [:base] (:log res)))
      (is (= [[:before nil] [:after nil]] @seen)))))

(deftest before-runs-newest-to-oldest-after-runs-oldest-to-newest
  (let [seen (atom [])
        note (fn [k] (fn [o] (swap! seen conj k) o))
        advised (-> (single-step-wf)
                    (wf/advice-add :t/step :before ::b1 (note :b1))
                    (wf/advice-add :t/step :before ::b2 (note :b2))
                    (wf/advice-add :t/step :after ::a1 (note :a1))
                    (wf/advice-add :t/step :after ::a2 (note :a2)))]
    (wf/run advised {})
    (is (= [:b2 :b1 :a1 :a2] @seen))))

(deftest throwing-advice-fails-the-step-through-the-contract
  (let [advised (-> (single-step-wf)
                    (wf/advice-add :t/step :before ::boom
                                   (fn [_] (throw (ex-info "advice boom" {})))))
        res (wf/run advised {})]
    (is (= 1 (:green/exit res)))
    (is (= "advice boom" (:green/err res)))
    (is (string? (:green/trace res)))))

(defn- two-step-wf []
  (wf/workflow {:start :t/a
               :wire-fn (fn [s _]
                          (case s
                            :t/a [(fn [o] (log o :a)) :t/b]
                            :t/b [(fn [o] (log o :b))]))}))

(deftest advice-add-all-applies-to-every-step
  (let [advised (-> (two-step-wf)
                    (wf/advice-add-all :filter-return ::tag #(log % :all)))]
    (is (= [:a :all :b :all] (:log (wf/run advised {}))))))

(deftest advice-add-all-interleaves-in-strict-add-order-with-per-step
  (let [advised (-> (two-step-wf)
                    (wf/advice-add-all :filter-return ::g1 #(log % :g1))
                    (wf/advice-add :t/a :filter-return ::pa #(log % :pa))
                    (wf/advice-add-all :filter-return ::g2 #(log % :g2)))
        res (wf/run advised {})]
    (testing ":t/a ran with both global entries plus its own, in add order"
      (is (= [:a :g1 :pa :g2] (subvec (:log res) 0 4))))
    (testing ":t/b, with no per-step advice, only saw the globals in add order"
      (is (= [:b :g1 :g2] (subvec (:log res) 4))))))

(deftest advice-remove-all-removes-the-global-entry
  (let [advised (-> (two-step-wf)
                    (wf/advice-add-all :filter-return ::g #(log % :g))
                    (wf/advice-remove-all ::g))]
    (is (= [:a :b] (:log (wf/run advised {}))))))

;; --- the while/until combinators -----------------------------------------

(defn- entry [how f seq]
  {:id (gensym "adv") :how how :fn f :seq seq :depth 0})

(deftest before-while-gates-the-inward-call
  (let [seen (atom [])
        base (fn [o] (swap! seen conj :base) (log o :base))]
    (testing "truthy advice lets the chain run; base sees the original opts"
      (reset! seen [])
      (let [f (advice/compose base [(entry :before-while
                                           (fn [_] (swap! seen conj :w) true) 0)])]
        (is (= [:base] (:log (f {}))))
        (is (= [:w :base] @seen))))
    (testing "nil advice short-circuits: nothing inward runs, result is nil"
      (reset! seen [])
      (let [f (advice/compose base [(entry :before-while
                                           (fn [_] (swap! seen conj :w) nil) 0)])]
        (is (nil? (f {})))
        (is (= [:w] @seen))))))

(deftest before-while-stack-runs-newest-to-oldest-and-stops-on-nil
  (let [seen (atom [])
        base (fn [o] (swap! seen conj :base) o)
        f (advice/compose base
                          [(entry :before-while (fn [_] (swap! seen conj :old) true) 0)
                           (entry :before-while (fn [_] (swap! seen conj :new) nil) 1)])]
    (is (nil? (f {})))
    (is (= [:new] @seen) "the newest gate is outermost and stops the chain")))

(deftest before-until-short-circuits-on-non-nil
  (testing "a non-nil advice return is the result; base never runs"
    (let [advised (-> (single-step-wf)
                      (wf/advice-add :t/step :before-until ::bu #(log % :bu)))]
      (is (= [:bu] (:log (wf/run advised {}))))))
  (testing "a nil advice return falls through to the chain"
    (let [advised (-> (single-step-wf)
                      (wf/advice-add :t/step :before-until ::bu (fn [_] nil)))]
      (is (= [:base] (:log (wf/run advised {})))))))

(deftest after-while-runs-oldest-to-newest-and-stops-on-nil
  (let [seen (atom [])
        base (fn [o] (swap! seen conj :base) o)
        f (advice/compose base
                          [(entry :after-while (fn [_] (swap! seen conj :old) nil) 0)
                           (entry :after-while (fn [_] (swap! seen conj :new) true) 1)])]
    (is (nil? (f {})))
    (is (= [:base :old] @seen)
        "base first, then oldest advice; its nil skips the newer one")))

(deftest after-while-and-after-until-use-green-exit-as-truth
  (testing ":after-while does not run after a failing step result"
    (let [seen (atom [])
          f (advice/compose (fn [o]
                              (swap! seen conj :base)
                              (assoc (log o :base) :green/exit 7))
                            [(entry :after-while
                                    (fn [o]
                                      (swap! seen conj :after)
                                      (assoc (log o :after) :green/exit 0))
                                    0)])
          ret (f {})]
      (is (= 7 (:green/exit ret)))
      (is (= [:base] (:log ret)))
      (is (= [:base] @seen))))
  (testing ":after-until falls through to advice after a failing step result"
    (let [seen (atom [])
          f (advice/compose (fn [o]
                              (swap! seen conj :base)
                              (assoc (log o :base) :green/exit 7))
                            [(entry :after-until
                                    (fn [o]
                                      (swap! seen conj :recover)
                                      (assoc (log o :recover) :green/exit 0))
                                    0)])
          ret (f {})]
      (is (= 0 (:green/exit ret)))
      (is (= [:recover] (:log ret)))
      (is (= [:base :recover] @seen)))))

(deftest after-until-supplies-a-result-when-the-chain-returns-nil
  (testing "compose level: advice only fires when the inward call is nil"
    (let [f (advice/compose (fn [_] nil)
                            [(entry :after-until (fn [o] (log o :fallback)) 0)])]
      (is (= [:fallback] (:log (f {})))))
    (let [f (advice/compose (fn [o] (log o :base))
                            [(entry :after-until (fn [o] (log o :fallback)) 0)])]
      (is (= [:base] (:log (f {}))) "a non-nil chain result short-circuits")))
  (testing "in a workflow: a before-while gate nils the step out and an
            outer after-until turns that into a real result"
    (let [advised (-> (single-step-wf)
                      (wf/advice-add :t/step :before-while ::gate (fn [_] nil))
                      (wf/advice-add :t/step :after-until ::fallback #(log % :fallback)))
          res (wf/run advised {})]
      (is (= [:fallback] (:log res)))
      (is (= 0 (:green/exit res))))))

(deftest after-until-stack-runs-oldest-to-newest
  (let [seen (atom [])
        f (advice/compose (fn [_] (swap! seen conj :base) nil)
                          [(entry :after-until (fn [_] (swap! seen conj :old) nil) 0)
                           (entry :after-until (fn [_] (swap! seen conj :new) :hit) 1)])]
    (is (= :hit (f {})))
    (is (= [:base :old :new] @seen))))

;; --- depth ----------------------------------------------------------------

(deftest depth-overrides-add-order
  (let [advised (-> (single-step-wf)
                    (wf/advice-add :t/step :filter-return ::outer #(log % :outer) {:depth -50})
                    (wf/advice-add :t/step :filter-return ::mid #(log % :mid))
                    (wf/advice-add :t/step :filter-return ::inner #(log % :inner) {:depth 50}))]
    (testing "lower depth is more outward, higher more inward, whatever the add order"
      (is (= [:base :inner :mid :outer] (:log (wf/run advised {})))))))

(deftest equal-depth-falls-back-to-newest-outermost
  (let [advised (-> (single-step-wf)
                    (wf/advice-add :t/step :filter-return ::a #(log % :a) {:depth -50})
                    (wf/advice-add :t/step :filter-return ::b #(log % :b) {:depth -50}))]
    (is (= [:base :a :b] (:log (wf/run advised {}))))))

(deftest depth-applies-across-per-step-and-all-steps-advice
  (let [advised (-> (two-step-wf)
                    (wf/advice-add-all :filter-return ::g #(log % :g))
                    (wf/advice-add :t/a :filter-return ::p #(log % :p) {:depth 100}))
        res (wf/run advised {})]
    (testing "depth 100 pushes the later-added per-step advice inside the global one"
      (is (= [:a :p :g] (subvec (:log res) 0 3))))
    (is (= [:b :g] (subvec (:log res) 3)))))

(deftest invalid-how-and-out-of-range-depth-are-rejected-at-add-time
  (is (thrown-with-msg? clojure.lang.ExceptionInfo #"unsupported advice combinator"
                        (wf/advice-add (single-step-wf) :t/step :befor ::typo identity)))
  (is (thrown-with-msg? clojure.lang.ExceptionInfo #":depth must be in -100..100"
                        (wf/advice-add (single-step-wf) :t/step :before ::deep identity
                                       {:depth 101}))))

;; --- cross-workflow inheritance (advice through wf/step) -------------------

(defn- embed
  "A parent workflow that runs :p/a, then `sub` as an embedded step, then :p/z."
  [sub]
  (wf/workflow {:start :p/a
                :wire-fn (fn [s _]
                           (case s
                             :p/a [(fn [o] (log o :p-a)) :p/sub]
                             :p/sub [(wf/step sub) :p/z]
                             :p/z [(fn [o] (log o :p-z))]))}))

(deftest parent-advice-reaches-an-embedded-step
  (let [sub (single-step-wf)
        parent (-> (embed sub)
                   (wf/advice-add :t/step :filter-return ::p #(log % :p)))]
    (is (= [:p-a :base :p :p-z] (:log (wf/run parent {}))))
    (testing "the child value is untouched — standalone runs stay bare"
      (is (= [:base] (:log (wf/run sub {})))))))

(deftest parent-override-redefines-an-embedded-step
  (let [parent (-> (embed (single-step-wf))
                   (wf/advice-add :t/step :override ::o #(log % :redefined)))]
    (is (= [:p-a :redefined :p-z] (:log (wf/run parent {}))))))

(deftest inherited-advice-is-outermost
  (let [sub (-> (single-step-wf)
                (wf/advice-add :t/step :filter-return ::c #(log % :child)))
        parent (-> (embed sub)
                   (wf/advice-add :t/step :filter-return ::p #(log % :parent)))]
    (is (= [:p-a :base :child :parent :p-z] (:log (wf/run parent {}))))))

(deftest same-id-parent-advice-replaces-the-childs
  (let [sub (-> (single-step-wf)
                (wf/advice-add :t/step :filter-return ::x #(log % :child-x)))
        parent (-> (embed sub)
                   (wf/advice-add :t/step :filter-return ::x #(log % :parent-x)))]
    (is (= [:p-a :base :parent-x :p-z] (:log (wf/run parent {}))))))

(deftest depth-overrides-inheritance-order
  (let [sub (-> (single-step-wf)
                (wf/advice-add :t/step :filter-return ::c #(log % :child) {:depth -50}))
        parent (-> (embed sub)
                   (wf/advice-add :t/step :filter-return ::p #(log % :parent)))]
    (testing "the child's depth -50 stays outside the parent's depth-0 advice"
      (is (= [:p-a :base :parent :child :p-z] (:log (wf/run parent {})))))))

(deftest advice-add-all-propagates-into-embeds
  (let [parent (-> (embed (single-step-wf))
                   (wf/advice-add-all :filter-return ::g #(log % :g)))]
    (testing "every inner step is wrapped, and so is the embed step itself"
      (is (= [:p-a :g :base :g :g :p-z :g] (:log (wf/run parent {})))))))

(deftest inheritance-survives-a-scoping-in-fn
  (let [parent (-> (wf/workflow {:start :p/sub
                                 :wire-fn (fn [_ _]
                                            [(wf/step (single-step-wf)
                                                      {:in (fn [o] {:n (:n o)})})])})
                   (wf/advice-add :t/step :filter-return ::p #(log % :p)))]
    (testing ":in built sub-opts from scratch; the engine re-stamped the registry"
      (is (= [:base :p] (:log (wf/run parent {:n 1})))))))

(deftest inheritance-is-transitive-through-nested-embeds
  (let [inner (single-step-wf)
        mid (wf/workflow {:start :m/sub :wire-fn (fn [_ _] [(wf/step inner)])})
        top (-> (wf/workflow {:start :top/sub :wire-fn (fn [_ _] [(wf/step mid)])})
                (wf/advice-add :t/step :filter-return ::p #(log % :top)))]
    (is (= [:base :top] (:log (wf/run top {}))))))

(deftest the-inherited-registry-does-not-leak-into-results
  (let [parent (-> (embed (single-step-wf))
                   (wf/advice-add :t/step :filter-return ::p identity))
        res (wf/run parent {})]
    (is (not (contains? res :green.workflow/inherited)))))

(deftest advice-plan-shows-the-cross-workflow-stack
  (let [sub (-> (single-step-wf)
                (wf/advice-add :t/step :before ::backend identity)
                (wf/advice-add :t/step :filter-return ::inner identity {:depth 50}))
        parent (-> (embed sub)
                   (wf/advice-add :t/step :before ::backend identity)
                   (wf/advice-add-all :around ::audit (fn [f o] (f o))))]
    (testing "outermost first, with provenance; the parent's ::backend replaced the child's"
      (is (= [[::audit :all 0] [::backend :step 0] [::inner :step 1]]
             (mapv (juxt :id :scope :level) (wf/advice-plan [parent sub] :t/step)))))
    (testing "a single workflow works too"
      (is (= [[::backend :step 0] [::inner :step 0]]
             (mapv (juxt :id :scope :level) (wf/advice-plan sub :t/step)))))))

(deftest advice-add-all-workflow-is-untouched-by-later-adds
  (let [base (two-step-wf)
        g1 (wf/advice-add-all base :filter-return ::g1 #(log % :g1))
        g2 (wf/advice-add-all g1 :filter-return ::g2 #(log % :g2))]
    (is (= [:a :g1 :b :g1] (:log (wf/run g1 {}))) "g1 is untouched by adding g2")
    (is (= [:a :g1 :g2 :b :g1 :g2] (:log (wf/run g2 {}))))))

(ns green.ansible-test
  "Playbook selection, recap parsing, and inventory rendering need no
  ansible binary — only `ansible-step` itself shells out."
  (:require [clojure.test :refer [deftest is testing]]
            [green.ansible :as ansible])
  (:import [java.nio.file Files]
           [java.nio.file.attribute FileAttribute]))

(defn- tmpdir []
  (str (Files/createTempDirectory "green-ansible" (make-array FileAttribute 0))))

(deftest playbook-follows-the-event
  (is (= "create.yml" (ansible/playbook {:green/event :create})))
  (is (= "create.yml" (ansible/playbook {:green/event :rotate}))
      "any non-delete event provisions")
  (is (= "delete.yml" (ansible/playbook {:green/event :delete})))
  (testing "partial overrides keep the other default"
    (is (= "teardown.yml"
           (ansible/playbook {:green/event :delete} {:delete "teardown.yml"})))
    (is (= "create.yml"
           (ansible/playbook {:green/event :create} {:delete "teardown.yml"})))))

(deftest recap-parses-per-host-counters
  (let [out (str "PLAY RECAP *********************************************\n"
                 "zk1                        : ok=7    changed=4    unreachable=0    failed=0    skipped=1    rescued=0    ignored=0\n"
                 "zk2                        : ok=7    changed=0    unreachable=0    failed=1    skipped=1    rescued=0    ignored=0\n")]
    (is (= {"zk1" {:ok 7 :changed 4 :unreachable 0 :failed 0
                   :skipped 1 :rescued 0 :ignored 0}
            "zk2" {:ok 7 :changed 0 :unreachable 0 :failed 1
                   :skipped 1 :rescued 0 :ignored 0}}
           (ansible/parse-recap out))))
  (is (= {} (ansible/parse-recap "no recap here"))))

(deftest inventory-ini-renders-groups-hosts-and-vars
  (is (= (str "[zookeeper]\n"
              "zk1 ansible_host=10.0.0.1 zk_id=1\n"
              "zk2 ansible_host=10.0.0.2 zk_id=2\n"
              "\n"
              "[zookeeper:vars]\n"
              "ansible_python_interpreter=/usr/bin/python3\n"
              "ansible_user=root\n")
         (ansible/inventory-ini
          {"zookeeper"
           {:hosts [{:name "zk1" :vars {:ansible_host "10.0.0.1" :zk_id 1}}
                    {:name "zk2" :vars {:ansible_host "10.0.0.2" :zk_id 2}}]
            :vars {:ansible_user "root"
                   :ansible_python_interpreter "/usr/bin/python3"}}}))))

(deftest inventory-ini-without-group-vars
  (is (= "[web]\nw1\n"
         (ansible/inventory-ini {:web {:hosts [{:name "w1"}]}}))))

(deftest inventory-advice-writes-the-file
  (let [dir (tmpdir)
        file (str dir "/hosts/inventory.ini")
        advice (ansible/inventory-advice
                (constantly file)
                (fn [opts]
                  {"zookeeper"
                   {:hosts (mapv (fn [{:keys [name ip]}]
                                   {:name name :vars {:ansible_host ip}})
                                 (:servers opts))}}))
        opts {:servers [{:name "zk1" :ip "10.0.0.1"}]}]
    (is (= opts (advice opts)) "before-advice passes opts through")
    (is (= "[zookeeper]\nzk1 ansible_host=10.0.0.1\n"
           (slurp file)))))

(ns green.cli-test
  (:require [clojure.test :refer [deftest is testing]]
            [green.cli :as cli]
            [green.workflow :as wf])
  (:import [java.io File]))

(defn- state-file [content]
  (let [f (File/createTempFile "green-state" ".edn")]
    (spit f content)
    (str f)))

(defn- probe-wf []
  (wf/workflow {:start :t/a
                :wire-fn (fn [s _]
                           (case s
                             :t/a [(fn [o] (update o :seen (fnil conj []) [:a (:green/event o) (:x o)]))
                                   :t/b]
                             :t/b [(fn [o] (update o :seen conj :b))]))}))

(deftest event-and-state-flow-into-the-workflow
  (let [res (cli/run-cli (probe-wf) ["create" "-f" (state-file "{:x 1}")])]
    (is (= 0 (:green/exit res)))
    (is (= [[:a :create 1] :b] (:seen res)))))

(deftest arbitrary-events-are-allowed
  (let [res (cli/run-cli (probe-wf) ["provision" "-f" (state-file "{:x 2}")])]
    (is (= [[:a :provision 2] :b] (:seen res)))))

(deftest slices-via-start-and-end
  (testing "--end is an inclusive boundary"
    (let [res (cli/run-cli (probe-wf)
                           ["create" "-f" (state-file "{:x 1}") "--end" "t/a"])]
      (is (= [[:a :create 1]] (:seen res)))))
  (testing "--start skips earlier steps"
    (let [res (cli/run-cli (probe-wf)
                           ["create" "-f" (state-file "{:x 1}") "--start" "t/b"])]
      (is (= [:b] (:seen res))))))

(deftest dry-run-flag-stamps-the-key
  (let [wf (wf/workflow {:start :t/a
                         :wire-fn (fn [_ _] [(fn [o] (assoc o :dry (:green/dry-run o)))])})]
    (is (true? (:dry (cli/run-cli wf ["create" "-f" (state-file "{}") "--dry-run"]))))
    (is (nil? (:dry (cli/run-cli wf ["create" "-f" (state-file "{}")]))))))

(deftest usage-errors-exit-2
  (testing "missing event"
    (let [res (cli/run-cli (probe-wf) [])]
      (is (= 2 (:green/exit res)))
      (is (re-find #"Usage" (:green/err res)))))
  (testing "missing state file"
    (let [res (cli/run-cli (probe-wf) ["create" "-f" "/nonexistent/green.edn"])]
      (is (= 2 (:green/exit res)))
      (is (re-find #"not found" (:green/err res))))))

(ns green.dry-run-test
  (:require [clojure.string :as str]
            [clojure.test :refer [deftest is testing]]
            [green.dry-run :as dry-run]
            [green.workflow :as wf]))

(defn- effect-wf [effects]
  (-> (wf/workflow {:start :t/a
                    :wire-fn (fn [s _]
                               (case s
                                 :t/a [(fn [o] (swap! effects conj :a) o) :t/b]
                                 :t/b [(fn [o] (swap! effects conj :b) o)]))})
      (dry-run/advise [:t/a :t/b])))

(deftest dry-run-skips-advised-steps
  (let [effects (atom [])
        out (with-out-str
              (is (= 0 (:green/exit (wf/run (effect-wf effects)
                                            {:green/event :create
                                             :green/dry-run true})))))]
    (is (= [] @effects) "no side effects ran")
    (is (str/includes? out "would run :t/a"))
    (is (str/includes? out "would run :t/b"))))

(deftest without-the-flag-steps-run-normally
  (let [effects (atom [])
        res (wf/run (effect-wf effects) {:green/event :create})]
    (is (= 0 (:green/exit res)))
    (is (= [:a :b] @effects))))

(deftest dry-run-advice-is-removable-per-step
  (let [effects (atom [])
        w (-> (effect-wf effects)
              (wf/advice-remove :t/b ::dry-run/skip))]
    (with-out-str (wf/run w {:green/event :create :green/dry-run true}))
    (testing ":t/a skipped, :t/b ran"
      (is (= [:b] @effects)))))

(ns green.progress-test
  (:require [clojure.string :as str]
            [clojure.test :refer [deftest is testing]]
            [green.progress :as progress]
            [green.workflow :as wf]))

(defn- simple-wf []
  (-> (wf/workflow {:start :t/a
                    :wire-fn (fn [s _]
                               (case s
                                 :t/a [(fn [o] o) :t/b]
                                 :t/b [(fn [o] o)]))})
      (progress/advise)))

(deftest progress-prints-step-names
  (let [out (with-out-str
              (wf/run (simple-wf) {:green/event :create}))]
    (is (str/includes? out ">>> :t/a (create)"))
    (is (str/includes? out "<<< :t/a"))
    (is (str/includes? out ">>> :t/b (create)"))
    (is (str/includes? out "<<< :t/b"))
    (is (str/includes? out "ms)") "elapsed time is printed")))

(deftest progress-prints-event-name
  (let [out (with-out-str
              (wf/run (simple-wf) {:green/event :delete}))]
    (is (str/includes? out ">>> :t/a (delete)"))))

(deftest progress-is-removable
  (let [w (wf/advice-remove-all (simple-wf) ::progress/progress)
        out (with-out-str (wf/run w {:green/event :create}))]
    (is (= "" out))))

(deftest progress-with-forks
  (let [w (-> (wf/workflow
                {:start :t/start
                 :wire-fn (fn [s _]
                            (case s
                              :t/start [(fn [o] o) :t/join]
                              :t/join [(fn [o] o)]))
                 :next-fn (fn [step dn o]
                            (if (= step :t/start)
                              [[:t/join (assoc o :branch :a)]
                               [:t/join (assoc o :branch :b)]]
                              (mapv (fn [s] [s o]) dn)))})
              (progress/advise))
        out (with-out-str (wf/run w {:green/event :create}))]
    (is (str/includes? out ">>> :t/start"))
    (is (str/includes? out ">>> :t/join"))))

(deftest green-step-is-set-in-opts
  (let [seen (atom [])
        w (wf/workflow {:start :t/a
                        :wire-fn (fn [s _]
                                   (case s
                                     :t/a [(fn [o] (swap! seen conj (:green/step o)) o) :t/b]
                                     :t/b [(fn [o] (swap! seen conj (:green/step o)) o)]))})
        res (wf/run w {:green/event :create})]
    (is (= [:t/a :t/b] @seen))
    (is (= 0 (:green/exit res)))))

(ns green.scaffold-test
  (:require [clojure.java.io :as io]
            [clojure.test :refer [deftest is testing]]
            [green.scaffold :as sc])
  (:import [java.nio.file Files]
           [java.nio.file.attribute FileAttribute]))

(defn- tmpdir []
  (str (Files/createTempDirectory "green-scaffold" (make-array FileAttribute 0))))

(deftest template-path-convention
  (is (= "zk/main.tf" (sc/template-path :zk/main.tf)))
  (is (= "my/app/zoo.cfg" (sc/template-path :my.app/zoo.cfg)))
  (is (= "plain.txt" (sc/template-path :plain.txt))))

(deftest scaffold-create-and-delete
  (let [dir (tmpdir)
        specs [{:template :greentest/hello.txt
                :target (str dir "/{{who}}/hello.txt")
                :data {:who "world" :name "green"}}]
        created (sc/scaffold {:green/event :create} specs)]
    (testing "create renders the template into the Selmer-rendered target"
      (is (= 0 (:green/exit created)))
      (is (= [(str dir "/world/hello.txt")] (:green.scaffold/written created)))
      (is (= "Hello green!\n" (slurp (str dir "/world/hello.txt")))))
    (testing "create is idempotent"
      (sc/scaffold {:green/event :create} specs)
      (is (= "Hello green!\n" (slurp (str dir "/world/hello.txt")))))
    (testing "delete removes targets and prunes emptied directories"
      (let [deleted (sc/scaffold {:green/event :delete} specs)]
        (is (= 0 (:green/exit deleted)))
        (is (not (.exists (io/file dir "world" "hello.txt"))))
        (is (not (.exists (io/file dir "world"))))))))

(deftest missing-template-throws-with-context
  (is (thrown-with-msg? Exception #"template not found"
        (sc/scaffold {:green/event :create}
                     [{:template :nope/missing.txt :target "/tmp/x" :data {}}]))))

(ns green.tofu-test
  "Backend advices need no tofu binary — they only write backend.tf."
  (:require [clojure.test :refer [deftest is]]
            [green.tofu :as tofu])
  (:import [java.nio.file Files]
           [java.nio.file.attribute FileAttribute]))

(defn- tmpdir []
  (str (Files/createTempDirectory "green-tofu" (make-array FileAttribute 0))))

(deftest local-backend-is-the-default
  (let [dir (tmpdir)
        advice (tofu/local-backend-advice (constantly dir))
        opts {:x 1}]
    (is (= opts (advice opts)) "before-advice passes opts through")
    (is (= "terraform {\n  backend \"local\" {\n  }\n}\n"
           (slurp (str dir "/backend.tf"))))))

(deftest s3-backend-advice-writes-attributes
  (let [dir (tmpdir)]
    ((tofu/s3-backend-advice (constantly dir)
                             {:bucket "my-state"
                              :key "green/node-1.tfstate"
                              :region "eu-west-1"
                              :encrypt true})
     {})
    (is (= (str "terraform {\n"
                "  backend \"s3\" {\n"
                "    bucket = \"my-state\"\n"
                "    encrypt = true\n"
                "    key = \"green/node-1.tfstate\"\n"
                "    region = \"eu-west-1\"\n"
                "  }\n"
                "}\n")
           (slurp (str dir "/backend.tf"))))))

(deftest backend-config-can-be-a-function-of-opts
  (let [dir (tmpdir)]
    ((tofu/gcs-backend-advice (constantly dir)
                              (fn [opts] {:bucket "state"
                                          :prefix (str "green/" (:node opts))}))
     {:node "n1"})
    (is (= (str "terraform {\n"
                "  backend \"gcs\" {\n"
                "    bucket = \"state\"\n"
                "    prefix = \"green/n1\"\n"
                "  }\n"
                "}\n")
           (slurp (str dir "/backend.tf"))))))

(ns green.workflow-test
  (:require [clojure.test :refer [deftest is testing]]
            [green.workflow :as wf]))

(defn- mark [k] (fn [o] (update o :seen (fnil conj []) k)))

(deftest linear-happy-path
  (let [w (wf/workflow {:start :t/a
                        :wire-fn (fn [s _]
                                   (case s
                                     :t/a [(mark :a) :t/b]
                                     :t/b [(mark :b) :t/c]
                                     :t/c [(mark :c)]))})
        res (wf/run w {})]
    (is (= [:a :b :c] (:seen res)))
    (is (= 0 (:green/exit res)))))

(deftest error-halts-without-next-fn
  (let [w (wf/workflow {:start :t/a
                        :wire-fn (fn [s _]
                                   (case s
                                     :t/a [(mark :a) :t/b]
                                     :t/b [(fn [o] (assoc o :green/exit 3)) :t/c]
                                     :t/c [(mark :c)]))})
        res (wf/run w {})]
    (is (= 3 (:green/exit res)))
    (is (= [:a] (:seen res)) ":t/c must not run")))

(deftest exception-becomes-exit-err-trace
  (let [w (wf/workflow {:start :t/a
                        :wire-fn (fn [s _]
                                   (case s
                                     :t/a [(fn [_] (throw (ex-info "boom" {})))
                                           :t/b]
                                     :t/b [(mark :b)]))})
        res (wf/run w {})]
    (is (= 1 (:green/exit res)))
    (is (= "boom" (:green/err res)))
    (is (string? (:green/trace res)))
    (is (nil? (:seen res)) ":t/b must not run")))

(deftest end-step-is-an-inclusive-slice-boundary
  (let [wire (fn [s _]
               (case s
                 :t/a [(mark :a) :t/b]
                 :t/b [(mark :b) :t/c]
                 :t/c [(mark :c)]))]
    (testing "end stops after running it"
      (is (= [:a :b]
             (:seen (wf/run (wf/workflow {:start :t/a :end :t/b :wire-fn wire}) {})))))
    (testing "start skips earlier steps"
      (is (= [:b :c]
             (:seen (wf/run (wf/workflow {:start :t/b :wire-fn wire}) {})))))))

(deftest wire-fn-can-select-an-event-specific-static-graph
  (let [w (wf/workflow
           {:start :t/start
            :wire-fn (fn [step run-opts]
                       (case [(:green/event run-opts) step]
                         [:create :t/start] [(mark :start) :t/node]
                         [:create :t/node] [(mark :node) :t/ansible]
                         [:create :t/ansible] [(mark :ansible)]
                         [:delete :t/start] [(mark :start) :t/ansible]
                         [:delete :t/ansible] [(mark :ansible) :t/node]
                         [:delete :t/node] [(mark :node)]))})]
    (is (= [:start :node :ansible]
           (:seen (wf/run w {:green/event :create}))))
    (is (= [:start :ansible :node]
           (:seen (wf/run w {:green/event :delete}))))))

(deftest wire-fn-uses-stable-run-opts-not-branch-opts
  (let [w (wf/workflow
           {:start :t/start
            :wire-fn (fn [step run-opts]
                       (case step
                         :t/start [identity :t/work]
                         :t/work [(mark :work)
                                  (if (= :branch (:route run-opts))
                                    :t/branch-done
                                    :t/run-done)]
                         :t/run-done [(mark :run-done)]
                         :t/branch-done [(mark :branch-done)]))
            :next-fn (fn [step default-next opts]
                       (if (= step :t/start)
                         [[:t/work (assoc opts :route :branch)]]
                         (mapv (fn [s] [s opts]) default-next)))})
        res (wf/run w {:route :run})]
    (is (= [:work :run-done] (:seen res)))
    (is (= :branch (:route res)) "the step still receives branch-local opts")))

(deftest next-fn-reroutes-errors
  (let [w (wf/workflow {:start :t/a
                        :wire-fn (fn [s _]
                                   (case s
                                     :t/a [(fn [o] (assoc o :green/exit 7)) :t/b]
                                     :t/b [(mark :b)]
                                     :t/cleanup [(fn [o] (-> o (assoc :green/exit 0)
                                                             ((mark :cleanup))))]))
                        :next-fn (fn [step dn opts]
                                   (cond
                                     (pos? (:green/exit opts 0)) (when-not (= step :t/cleanup)
                                                                   [[:t/cleanup opts]])
                                     :else (map (fn [s] [s opts]) dn)))})
        res (wf/run w {})]
    (is (= [:cleanup] (:seen res)))
    (is (= 0 (:green/exit res)))))

(deftest next-fn-nil-terminates
  (let [w (wf/workflow {:start :t/a
                        :wire-fn (fn [s _]
                                   (case s
                                     :t/a [(mark :a) :t/b]
                                     :t/b [(mark :b)]))
                        :next-fn (fn [_ _ _] nil)})
        res (wf/run w {})]
    (is (= [:a] (:seen res)))))

(deftest static-fork-and-join
  ;; :t/a forks to :t/b and :t/c; :t/c has a longer chain (:t/c -> :t/c2);
  ;; both arrive at :t/d, which must run once with both branches collected.
  (let [w (wf/workflow {:start :t/a
                        :wire-fn (fn [s _]
                                   (case s
                                     :t/a [(mark :a) :t/b :t/c]
                                     :t/b [(mark :b) :t/d]
                                     :t/c [(mark :c) :t/c2]
                                     :t/c2 [(mark :c2) :t/d]
                                     :t/d [(fn [o]
                                             (assoc o :joined
                                                    (mapv :seen (:green/branches o))))]))})
        res (wf/run w {})]
    (is (= 0 (:green/exit res)))
    (is (= 2 (count (:green/branches res))))
    (testing "join waited for the longer branch"
      (is (= #{[:a :b] [:a :c :c2]} (set (:joined res)))))
    (testing "join base opts come from the fork point"
      (is (= [:a] (:seen res))
          "join step's own opts derive from fork opts, not a branch"))))

(deftest dynamic-fan-out-and-join
  (let [w (wf/workflow {:start :t/fan
                        :wire-fn (fn [s _]
                                   (case s
                                     :t/fan [(mark :fan) :t/work]
                                     :t/work [(fn [o] (assoc o :done (:n o)))
                                              :t/join]
                                     :t/join [(fn [o]
                                                (assoc o :collected
                                                       (set (map :done (:green/branches o)))))]))
                        :next-fn (fn [step dn opts]
                                   (cond
                                     (pos? (:green/exit opts 0)) []
                                     (= step :t/fan) (for [n [1 2 3]]
                                                       [:t/work (assoc opts :n n)])
                                     :else (map (fn [s] [s opts]) dn)))})
        res (wf/run w {})]
    (is (= 0 (:green/exit res)))
    (is (= 3 (count (:green/branches res))))
    (is (= #{1 2 3} (:collected res)))))

(deftest branch-failure-skips-join-and-propagates-worst-exit
  (let [w (wf/workflow {:start :t/a
                        :wire-fn (fn [s _]
                                   (case s
                                     :t/a [(mark :a) :t/ok :t/bad]
                                     :t/ok [(mark :ok) :t/d]
                                     :t/bad [(fn [o] (assoc o :green/exit 5
                                                            :green/err "bad branch"))
                                             :t/d]
                                     :t/d [(mark :join-ran)]))})
        res (wf/run w {})]
    (is (= 5 (:green/exit res)))
    (is (= "bad branch" (:green/err res)))
    (is (not-any? #(= % :join-ran) (:seen res [])) "join must be skipped")
    (is (= 2 (count (:green/branches res))) "both branches finished")))

(deftest workflows-compose-as-steps
  (let [sub (wf/workflow {:start :s/a
                          :wire-fn (fn [s _]
                                     (case s
                                       :s/a [(mark :sub-a) :s/b]
                                       :s/b [(mark :sub-b)]))})
        parent (wf/workflow {:start :p/a
                             :wire-fn (fn [s _]
                                        (case s
                                          :p/a [(mark :p-a) :p/sub]
                                          :p/sub [(wf/step sub) :p/z]
                                          :p/z [(mark :p-z)]))})
        res (wf/run parent {})]
    (is (= 0 (:green/exit res)))
    (is (= [:p-a :sub-a :sub-b :p-z] (:seen res)))))

(deftest sub-workflow-failure-halts-the-parent
  (let [sub (wf/workflow {:start :s/a
                          :wire-fn (fn [_ _] [(fn [o] (assoc o :green/exit 4
                                                           :green/err "sub failed"))])})
        parent (wf/workflow {:start :p/sub
                             :wire-fn (fn [s _]
                                        (case s
                                          :p/sub [(wf/step sub) :p/z]
                                          :p/z [(mark :p-z)]))})
        res (wf/run parent {})]
    (is (= 4 (:green/exit res)))
    (is (= "sub failed" (:green/err res)))
    (is (nil? (:seen res)) ":p/z must not run")))

(deftest step-in-out-scope-the-sub-workflow
  (let [sub (wf/workflow {:start :s/a
                          :wire-fn (fn [_ _] [(fn [o] (assoc o :result (* 2 (:n o))))])})
        parent (wf/workflow {:start :p/sub
                             :wire-fn (fn [_ _]
                                        [(wf/step sub
                                                  {:in (fn [o] {:green/event (:green/event o)
                                                                :n (:parent-n o)})
                                                   :out (fn [o r] (assoc o :doubled (:result r)))})])})
        res (wf/run parent {:green/event :create :parent-n 21 :keep-me :yes})]
    (is (= 42 (:doubled res)))
    (is (= :yes (:keep-me res)) ":out preserved the parent opts")))

(deftest parallel-branches-actually-run-concurrently
  (let [in-flight (atom 0)
        peak (atom 0)
        slow (fn [o]
               (swap! peak max (swap! in-flight inc))
               (Thread/sleep 50)
               (swap! in-flight dec)
               o)
        w (wf/workflow {:start :t/a
                        :wire-fn (fn [s _]
                                   (case s
                                     :t/a [identity :t/x :t/y :t/z]
                                     :t/x [slow :t/d]
                                     :t/y [slow :t/d]
                                     :t/z [slow :t/d]
                                     :t/d [identity]))})
        res (wf/run w {:green/exit 0})]
    (is (= 0 (:green/exit res)))
    (is (< 1 @peak) "branches overlapped in time")))

(ns green.zookeeper-test
  "Flagship end-to-end test: a fake ZooKeeper cluster. Real Selmer renders,
  real `tofu apply`/`destroy` — but the HCL contains only locals and outputs,
  so nothing real is created and no credentials are needed."
  (:require [clojure.java.io :as io]
            [clojure.java.shell :as sh]
            [clojure.string :as str]
            [clojure.test :refer [deftest is testing]]
            [green.scaffold :as sc]
            [green.tofu :as tofu]
            [green.workflow :as wf])
  (:import [java.nio.file Files]
           [java.nio.file.attribute FileAttribute]))

(defn- tofu-available? []
  (try (zero? (:exit (sh/sh "tofu" "version")))
       (catch Exception _ false)))

(defn- tmpdir []
  (str (Files/createTempDirectory "green-zk" (make-array FileAttribute 0))))

(defn- node-dir [opts node]
  (str (:zk/workdir opts) "/nodes/" (:id node)))

;; --- steps -----------------------------------------------------------------

(defn start-step [opts] opts)

(defn node-step
  "Scaffold one node's main.tf and drive tofu over it. On delete, destroy
  before removing the files tofu still needs."
  [opts]
  (let [node (:zk/node opts)
        dir (node-dir opts node)
        specs [{:template :zk/main.tf
                :target (str dir "/main.tf")
                :data {:node node}}]]
    (if (= :delete (:green/event opts))
      (let [opts (tofu/tofu-step opts {:dir dir})]
        (if (pos? (:green/exit opts 0)) opts (sc/scaffold opts specs)))
      (let [opts (sc/scaffold opts specs)]
        (tofu/tofu-step opts {:dir dir})))))

(defn zoo-cfg-step
  "Join step: render zoo.cfg (listing every member) into each server's
  directory. On create the member list comes from the branches' observed
  tofu outputs; on delete the desired state names the targets to remove."
  [opts]
  (let [nodes (:zk/servers opts)
        servers (if (= :delete (:green/event opts))
                  nodes
                  (->> (:green/branches opts)
                       (map :tofu/outputs)
                       (sort-by :id)))
        specs (for [n nodes]
                {:template :zk/zoo.cfg
                 :target (str (:zk/workdir opts) "/nodes/" (:id n) "/zoo.cfg")
                 :data {:servers servers}})]
    (sc/scaffold opts specs)))

;; --- workflow ----------------------------------------------------------------

(defn wire-fn [step _]
  (case step
    :zk/start   [start-step :zk/node]
    :zk/node    [node-step :zk/zoo-cfg]
    :zk/zoo-cfg [zoo-cfg-step]))

(defn next-fn [step default-next opts]
  (cond
    (pos? (:green/exit opts 0)) []
    ;; fan out: one :zk/node branch per server, in parallel
    (= step :zk/start) (mapv (fn [n] [:zk/node (assoc opts :zk/node n)])
                             (:zk/servers opts))
    :else (mapv (fn [s] [s opts]) default-next)))

(def cluster-wf
  (-> (wf/workflow {:start :zk/start :wire-fn wire-fn :next-fn next-fn})
      ;; the backend is not hardwired: the local filesystem backend is
      ;; injected with the advice facility
      (wf/advice-add :zk/node :before ::backend
                     (tofu/local-backend-advice #(node-dir % (:zk/node %))))))

(def desired-state
  {:zk/servers [{:id 1 :name "zk1" :ip "10.0.0.1"}
                {:id 2 :name "zk2" :ip "10.0.0.2"}
                {:id 3 :name "zk3" :ip "10.0.0.3"}]})

;; --- composition: two clusters from the same workflow ------------------------
;; `wf/step` turns cluster-wf into an ordinary step; the parent fans it out
;; once per cluster (in parallel), scoping each run with :in.

(def two-clusters-wf
  (wf/workflow
   {:start :clusters/start
    :wire-fn (fn [step _]
               (case step
                 :clusters/start [start-step :clusters/cluster]
                 :clusters/cluster [(wf/step cluster-wf
                                             {:in (fn [opts]
                                                    (let [c (:zk/cluster opts)]
                                                      (assoc opts
                                                             :zk/servers (:servers c)
                                                             :zk/workdir (str (:zk/workdir opts)
                                                                              "/" (:name c)))))})
                                    :clusters/report]
                 :clusters/report [(fn [opts]
                                     (assoc opts :clusters/reported
                                            (set (map :zk/workdir (:green/branches opts)))))]))
    :next-fn (fn [step default-next opts]
               (cond
                 (pos? (:green/exit opts 0)) []
                 (= step :clusters/start) (mapv (fn [c] [:clusters/cluster (assoc opts :zk/cluster c)])
                                                (:zk/clusters opts))
                 :else (mapv (fn [s] [s opts]) default-next)))}))

(def two-clusters-state
  {:zk/clusters [{:name "alpha"
                  :servers [{:id 1 :name "zk1" :ip "10.0.1.1"}
                            {:id 2 :name "zk2" :ip "10.0.1.2"}]}
                 {:name "beta"
                  :servers [{:id 1 :name "zk1" :ip "10.0.2.1"}
                            {:id 2 :name "zk2" :ip "10.0.2.2"}]}]})

(deftest two-zookeeper-clusters
  (if-not (tofu-available?)
    (println "SKIP green.zookeeper-test/two-zookeeper-clusters: tofu not on PATH")
    (let [work (tmpdir)
          state (assoc two-clusters-state :zk/workdir work)]

      (testing "create: the cluster workflow runs twice, in parallel"
        (let [res (wf/run two-clusters-wf (assoc state :green/event :create))]
          (is (= 0 (:green/exit res)) (str (:green/err res)))
          (is (= 2 (count (:green/branches res))))
          (is (= #{(str work "/alpha") (str work "/beta")} (:clusters/reported res)))
          (doseq [{cname :name servers :servers} (:zk/clusters state)
                  {:keys [id]} servers]
            (let [dir (str work "/" cname "/nodes/" id)]
              (is (.exists (io/file dir "main.tf")) dir)
              (is (.exists (io/file dir "terraform.tfstate")) dir)
              (let [cfg (slurp (io/file dir "zoo.cfg"))]
                (doseq [{:keys [id ip]} servers]
                  (is (str/includes? cfg (str "server." id "=" ip ":2888:3888"))
                      "zoo.cfg lists this cluster's members")))))
          (testing "clusters stay isolated"
            (is (not (str/includes? (slurp (io/file (str work "/alpha/nodes/1") "zoo.cfg"))
                                    "10.0.2."))
                "alpha's zoo.cfg must not list beta's nodes"))))

      (testing "delete tears down both clusters"
        (let [res (wf/run two-clusters-wf (assoc state :green/event :delete))]
          (is (= 0 (:green/exit res)) (str (:green/err res)))
          (doseq [{cname :name servers :servers} (:zk/clusters state)
                  {:keys [id]} servers]
            (let [dir (str work "/" cname "/nodes/" id)]
              (is (not (.exists (io/file dir "main.tf"))))
              (is (not (.exists (io/file dir "zoo.cfg")))))))))))

(deftest zookeeper-fake-cluster
  (if-not (tofu-available?)
    (println "SKIP green.zookeeper-test: tofu not on PATH")
    (let [state (assoc desired-state :zk/workdir (tmpdir))
          work (:zk/workdir state)]

      (testing "create: fan-out, parallel tofu applies, join renders zoo.cfg"
        (let [res (wf/run cluster-wf (assoc state :green/event :create))]
          (is (= 0 (:green/exit res)) (str (:green/err res)))
          (is (= 3 (count (:green/branches res))))
          (testing "observed outputs flowed back into each branch"
            (is (= #{"zk1" "zk2" "zk3"}
                   (set (map (comp :name :tofu/outputs) (:green/branches res))))))
          (doseq [{:keys [id]} (:zk/servers state)]
            (let [dir (str work "/nodes/" id)]
              (is (.exists (io/file dir "main.tf")))
              (is (.exists (io/file dir "backend.tf")) "advice wrote the backend")
              (is (.exists (io/file dir "terraform.tfstate")) "local backend state")
              (let [cfg (slurp (io/file dir "zoo.cfg"))]
                (doseq [{:keys [id ip]} (:zk/servers state)]
                  (is (str/includes? cfg (str "server." id "=" ip ":2888:3888"))
                      "every zoo.cfg lists every member")))))))

      (testing "create is idempotent"
        (is (= 0 (:green/exit (wf/run cluster-wf (assoc state :green/event :create))))))

      (testing "delete: destroys each node and removes the generated files"
        (let [res (wf/run cluster-wf (assoc state :green/event :delete))]
          (is (= 0 (:green/exit res)) (str (:green/err res)))
          (doseq [{:keys [id]} (:zk/servers state)]
            (let [dir (str work "/nodes/" id)]
              (is (not (.exists (io/file dir "main.tf"))))
              (is (not (.exists (io/file dir "zoo.cfg")))))))))))

#!/usr/bin/env bb
;; Self-contained green CLI for a fake ZooKeeper cluster.
;; Run from this directory:   ./green create   |   ./green delete
(require '[babashka.fs :as fs])

(def script-dir
  (if (= "NO_SOURCE_PATH" *file*)
    (fs/cwd)
    (fs/parent (fs/canonicalize *file*))))

(def repo-root (fs/canonicalize (fs/path script-dir "../..")))

;; Under babashka, pull in green + this example's resources at runtime.
;; Under the JVM (e.g. cider-jack-in with clojure-cli) these namespaces don't
;; exist and the sibling deps.edn supplies the classpath instead, so the whole
;; block is skipped. requiring-resolve keeps the JVM compiler from ever seeing
;; the babashka-only vars.
(when (System/getProperty "babashka.version")
  ((requiring-resolve 'babashka.deps/add-deps)
   {:deps {'io.github.amiorin/green {:local/root (str repo-root)}}})
  ((requiring-resolve 'babashka.classpath/add-classpath)
   (str (fs/path script-dir "resources"))))

(require '[green.cli :as cli]
         '[green.dry-run :as dry-run]
         '[green.scaffold :as sc]
         '[green.tofu :as tofu]
         '[green.workflow :as wf])

(defn example-path [path]
  (when path
    (let [p (fs/path path)]
      (str (if (fs/absolute? p) p (fs/path script-dir p))))))

(defn workdir [opts]
  (example-path (:zk/workdir opts)))

(defn node-dir [opts node]
  (str (workdir opts) "/nodes/" (:id node)))

(defn start-step [opts] opts)

(defn node-step [opts]
  (let [node (:zk/node opts)
        dir (node-dir opts node)
        specs [{:template :zk/main.tf
                :target (str dir "/main.tf")
                :data {:node node}}]]
    (if (= :delete (:green/event opts))
      (let [opts (tofu/tofu-step opts {:dir dir})]
        (if (pos? (:green/exit opts 0)) opts (sc/scaffold opts specs)))
      (let [opts (sc/scaffold opts specs)]
        (tofu/tofu-step opts {:dir dir})))))

(defn zoo-cfg-step [opts]
  (let [nodes (:zk/servers opts)
        servers (if (= :delete (:green/event opts))
                  nodes
                  (->> (:green/branches opts) (map :tofu/outputs) (sort-by :id)))
        specs (for [n nodes]
                {:template :zk/zoo.cfg
                 :target (str (workdir opts) "/nodes/" (:id n) "/zoo.cfg")
                 :data {:servers servers}})]
    (sc/scaffold opts specs)))

(defn wire-fn [step _]
  (case step
    :zk/start   [start-step :zk/node]
    :zk/node    [node-step :zk/zoo-cfg]
    :zk/zoo-cfg [zoo-cfg-step]))

(defn next-fn [step default-next opts]
  (cond
    (pos? (:green/exit opts 0)) []
    (= step :zk/start) (mapv (fn [n] [:zk/node (assoc opts :zk/node n)])
                             (:zk/servers opts))
    :else (mapv (fn [s] [s opts]) default-next)))

(def workflow
  (-> (wf/workflow {:start :zk/start :wire-fn wire-fn :next-fn next-fn})
      (wf/advice-add :zk/node :before ::backend
                     (tofu/local-backend-advice #(node-dir % (:zk/node %))))
      ;; ./green create --dry-run prints what would run instead of running it
      (dry-run/advise [:zk/node :zk/zoo-cfg])))

(defn file-arg? [arg]
  (or (= "-f" arg)
      (= "--file" arg)
      (.startsWith (str arg) "--file=")))

(defn default-args [args]
  (let [args (vec args)]
    (cond-> args
      (not-any? file-arg? args)
      (conj "-f" (str (fs/path script-dir "green.edn"))))))

(defn launched-as-script? []
  (when-let [bb-file (System/getProperty "babashka.file")]
    (= (str (fs/canonicalize bb-file))
       (str (fs/canonicalize *file*)))))

(defn run
  "REPL-friendly entrypoint: runs without calling System/exit."
  [& args]
  (cli/run-cli workflow (default-args args)))

(defn -main [& args]
  (cli/exec workflow (default-args args)))

(when (launched-as-script?)
  (apply -main *command-line-args*))

#!/usr/bin/env bb
;; Two fake ZooKeeper clusters from ONE cluster workflow, composed with
;; wf/step. Run from this directory:   ./green create   |   ./green delete
(require '[babashka.fs :as fs])

(def script-dir
  (if (= "NO_SOURCE_PATH" *file*)
    (fs/cwd)
    (fs/parent (fs/canonicalize *file*))))

(def repo-root (fs/canonicalize (fs/path script-dir "../..")))

;; Under babashka, pull in green + this example's resources at runtime.
;; Under the JVM (e.g. cider-jack-in with clojure-cli) these namespaces don't
;; exist and the sibling deps.edn supplies the classpath instead, so the whole
;; block is skipped. requiring-resolve keeps the JVM compiler from ever seeing
;; the babashka-only vars.
(when (System/getProperty "babashka.version")
  ((requiring-resolve 'babashka.deps/add-deps)
   {:deps {'io.github.amiorin/green {:local/root (str repo-root)}}})
  ((requiring-resolve 'babashka.classpath/add-classpath)
   (str (fs/path script-dir "resources"))))

(require '[green.cli :as cli]
         '[green.dry-run :as dry-run]
         '[green.scaffold :as sc]
         '[green.tofu :as tofu]
         '[green.workflow :as wf])

;; --- the single-cluster workflow (same shape as examples/zookeeper) ---------

(defn example-path [path]
  (when path
    (let [p (fs/path path)]
      (str (if (fs/absolute? p) p (fs/path script-dir p))))))

(defn workdir [opts]
  (example-path (:zk/workdir opts)))

(defn node-dir [opts node]
  (str (workdir opts) "/nodes/" (:id node)))

(defn start-step [opts] opts)

(defn logln [& xs]
  (locking *out*
    (apply println xs)
    (flush)))

(defn node-step [opts]
  (let [node (:zk/node opts)
        dir (node-dir opts node)
        specs [{:template :zk/main.tf
                :target (str dir "/main.tf")
                :data {:node node}}]]
    (if (= :delete (:green/event opts))
      (let [opts (tofu/tofu-step opts {:dir dir})]
        (if (pos? (:green/exit opts 0)) opts (sc/scaffold opts specs)))
      (let [opts (sc/scaffold opts specs)]
        (tofu/tofu-step opts {:dir dir})))))

(defn zoo-cfg-step [opts]
  (let [nodes (:zk/servers opts)
        servers (if (= :delete (:green/event opts))
                  nodes
                  (->> (:green/branches opts) (map :tofu/outputs) (sort-by :id)))
        specs (for [n nodes]
                {:template :zk/zoo.cfg
                 :target (str (workdir opts) "/nodes/" (:id n) "/zoo.cfg")
                 :data {:servers servers}})]
    (sc/scaffold opts specs)))

(defn cluster-wire-fn [step _]
  (case step
    :zk/start   [start-step :zk/node]
    :zk/node    [node-step :zk/zoo-cfg]
    :zk/zoo-cfg [zoo-cfg-step]))

(defn cluster-next-fn [step default-next opts]
  (cond
    (pos? (:green/exit opts 0)) []
    (= step :zk/start) (mapv (fn [n] [:zk/node (assoc opts :zk/node n)])
                             (:zk/servers opts))
    :else (mapv (fn [s] [s opts]) default-next)))

(def cluster-wf
  (-> (wf/workflow {:start :zk/start :wire-fn cluster-wire-fn :next-fn cluster-next-fn})
      ;; the child ships its default backend; a parent can replace it by
      ;; re-adding advice under the same id (::backend), or redefine
      ;; :zk/node wholesale with an :override advice
      (wf/advice-add :zk/node :before ::backend
                     (tofu/local-backend-advice #(node-dir % (:zk/node %))))))

;; --- the higher-level workflow: one cluster-wf run per cluster --------------

(defn clusters-wire-fn [step _]
  (case step
    :clusters/start [start-step :clusters/cluster]
    :clusters/cluster [(wf/step cluster-wf
                                {:in (fn [opts]
                                       (let [c (:zk/cluster opts)]
                                         (assoc opts
                                                :zk/servers (:servers c)
                                                :zk/workdir (str (:zk/workdir opts)
                                                                 "/" (:name c)))))})
                       :clusters/report]
    :clusters/report [(fn [opts]
                        (doseq [b (:green/branches opts)]
                          (logln "cluster ready:" (:zk/workdir b)))
                        opts)]))

(defn clusters-next-fn [step default-next opts]
  (cond
    (pos? (:green/exit opts 0)) []
    (= step :clusters/start) (mapv (fn [c] [:clusters/cluster (assoc opts :zk/cluster c)])
                                   (:zk/clusters opts))
    :else (mapv (fn [s] [s opts]) default-next)))

(def workflow
  (-> (wf/workflow {:start :clusters/start
                    :wire-fn clusters-wire-fn
                    :next-fn clusters-next-fn})
      ;; cross-workflow advice: :zk/node and :zk/zoo-cfg live inside the
      ;; embedded cluster-wf, but advice is inherited through wf/step —
      ;; flat step names reach any depth, parent advice is outermost
      (wf/advice-add :zk/node :before ::audit
                     (fn [opts]
                       (logln "node" (get-in opts [:zk/node :name])
                              "of cluster" (get-in opts [:zk/cluster :name]))))
      ;; dry-run for the whole composition, declared once at the top;
      ;; inspect the resulting stack with e.g.
      ;;   (wf/advice-plan [workflow cluster-wf] :zk/node)
      (dry-run/advise [:zk/node :zk/zoo-cfg])))

(defn file-arg? [arg]
  (or (= "-f" arg)
      (= "--file" arg)
      (.startsWith (str arg) "--file=")))

(defn default-args [args]
  (let [args (vec args)]
    (cond-> args
      (not-any? file-arg? args)
      (conj "-f" (str (fs/path script-dir "green.edn"))))))

(defn launched-as-script? []
  (when-let [bb-file (System/getProperty "babashka.file")]
    (= (str (fs/canonicalize bb-file))
       (str (fs/canonicalize *file*)))))

(defn run
  "REPL-friendly entrypoint: runs without calling System/exit."
  [& args]
  (cli/run-cli workflow (default-args args)))

(defn -main [& args]
  (cli/exec workflow (default-args args)))

(when (launched-as-script?)
  (apply -main *command-line-args*))

#!/usr/bin/env bb
;; Self-contained green CLI for a single-machine PaaS in the style of
;; Basecamp ONCE. One VPS runs a dockerized website operated by ONCE; green
;; provisions the infra with OpenTofu (fake HCL, locals/output only) and
;; scaffolds the Ansible config that would configure the box.
;;
;;   ./green create   |   ./green create --dry-run   |   ./green delete
;;
;; Graph:
;;   compute -+                              +-> ansible-local  (~/.ssh/config)
;;            +-> dns -> smtp-post ----------+
;;   smtp ----+                              +-> ansible-remote (playbook)
;;
;; See SPEC.md for the full design.
(require '[babashka.fs :as fs])

(def script-dir
  (if (= "NO_SOURCE_PATH" *file*)
    (fs/cwd)
    (fs/parent (fs/canonicalize *file*))))

(def repo-root (fs/canonicalize (fs/path script-dir "../..")))

;; Under babashka, pull in green + this example's resources at runtime.
;; Under the JVM (e.g. cider-jack-in with clojure-cli) these namespaces don't
;; exist and the sibling deps.edn supplies the classpath instead, so the whole
;; block is skipped. requiring-resolve keeps the JVM compiler from ever seeing
;; the babashka-only vars.
(when (System/getProperty "babashka.version")
  ((requiring-resolve 'babashka.deps/add-deps)
   {:deps {'io.github.amiorin/green {:local/root (str repo-root)}}})
  ((requiring-resolve 'babashka.classpath/add-classpath)
   (str (fs/path script-dir "resources"))))

(require '[clojure.java.io :as io]
         '[clojure.string :as str]
         '[green.cli :as cli]
         '[green.dry-run :as dry-run]
         '[green.scaffold :as sc]
         '[green.tofu :as tofu]
         '[green.workflow :as wf])

;; --- paths: one isolated tofu working dir per step --------------------------
;; Dir names are the generic resource they hold (server, smtp), not the step.

(defn example-path [path]
  (when path
    (let [p (fs/path path)]
      (str (if (fs/absolute? p) p (fs/path script-dir p))))))

(defn workdir [opts]
  (example-path (:once/workdir opts)))

(defn step-dir [opts nm] (str (workdir opts) "/" nm))

(defn server-dir [opts] (step-dir opts "server"))

(defn smtp-dir [opts] (step-dir opts "smtp"))

(defn dns-dir [opts] (step-dir opts "dns"))

(defn smtp-post-dir [opts] (step-dir opts "smtp-post"))

;; --- domain discovery from :once/website ------------------------------------

(defn apex
  "example.com from chat.example.com (last two labels)."
  [hostname]
  (->> (str/split hostname #"\.") (take-last 2) (str/join ".")))

(defn sending-domain
  "The smtp identity is always notifications.<domain>.<tld>, derived from the
  website hostnames' apex."
  [opts]
  (str "notifications." (apex (:hostname (first (:once/website opts))))))

(defn keywordize [m]
  (into {} (map (fn [[k v]] [(keyword k) v])) m))

;; --- provider-swap advice (headline) ----------------------------------------
;; Exactly the tofu.clj backend-as-advice idiom: a :before advice that
;; scaffolds the provider-specific main.tf before `compute` runs. Swap clouds
;; by re-adding under the same ::provider id — downstream steps never learn
;; which cloud produced the host.

(defn provider-advice [template dir-fn]
  (fn [opts]
    (let [dir (io/file (dir-fn opts))]
      (.mkdirs dir)
      (spit (io/file dir "main.tf")
            (sc/render-template template {:host (:once/host opts)
                                          :ssh  (:once/ssh opts)})))
    opts))

(def digitalocean-provider (provider-advice :once.do/main.tf server-dir))

(def oci-provider (provider-advice :once.oci/main.tf server-dir))

;; --- tofu steps -------------------------------------------------------------

(defn start-step [opts] opts)

;; compute is a bare tofu run: its main.tf is written by the ::provider
;; advice and its backend.tf by the ::backend advice, both :before.
(defn compute-step [opts]
  (tofu/tofu-step opts {:dir (server-dir opts)}))

(defn tofu-with-spec
  "Scaffold main.tf then apply (create); or destroy then remove main.tf
  (delete). The zookeeper node-step pattern."
  [opts dir specs]
  (if (= :delete (:green/event opts))
    (let [opts (tofu/tofu-step opts {:dir dir})]
      (if (pos? (:green/exit opts 0)) opts (sc/scaffold opts specs)))
    (let [opts (sc/scaffold opts specs)]
      (tofu/tofu-step opts {:dir dir}))))

(defn smtp-step [opts]
  (let [dir (smtp-dir opts)
        sub (sending-domain opts)
        specs [{:template :once.smtp/main.tf
                :target (str dir "/main.tf")
                :data {:subdomain sub
                       :id (str "snd-" (str/replace (apex (:hostname (first (:once/website opts)))) "." "-"))
                       :username "resend"
                       :password "mock-smtp-secret"}}]]
    (tofu-with-spec opts dir specs)))

;; dns joins compute and smtp: it needs compute.:ip and smtp.:records. It also
;; carries the two provisioning outputs forward (:once/server, :once/smtp) so
;; smtp-post and the ansible steps can read them without :green/branches.
(defn dns-step [opts]
  (let [dir (dns-dir opts)
        outs (map :tofu/outputs (:green/branches opts))
        compute-out (some #(when (:ip %) %) outs)
        smtp-out (some #(when (:records %) %) outs)
        specs [{:template :once.cloudflare/main.tf
                :target (str dir "/main.tf")
                :data {:ip (:ip compute-out)
                       :hostnames (map :hostname (:once/website opts))
                       :email (map keywordize (:records smtp-out))}}]
        opts (assoc opts :once/server compute-out :once/smtp smtp-out)]
    (tofu-with-spec opts dir specs)))

(defn smtp-post-step [opts]
  (let [dir (smtp-post-dir opts)
        specs [{:template :once.smtp-post/main.tf
                :target (str dir "/main.tf")
                :data {:id (:id (:once/smtp opts))
                       :subdomain (sending-domain opts)}}]]
    (tofu-with-spec opts dir specs)))

;; --- scaffold-only steps (no command) ---------------------------------------

;; ansible-local configures the OPERATOR's machine: a ~/.ssh/config Host entry
;; for the new VPS, rendered as a fragment to Include.
(defn ansible-local-step [opts]
  (let [srv (:once/server opts)
        specs [{:template :once/ssh-config
                :target (str (workdir opts) "/ansible-local/config")
                :data {:name (:name srv)
                       :ip (:ip srv)
                       :sudoer (:sudoer srv)
                       :identity "~/.ssh/once_compute"}}]]
    (sc/scaffold opts specs)))

;; ansible-remote configures the BOX: the playbook (Docker + ONCE + sites).
(defn ansible-remote-step [opts]
  (let [specs [{:template :once/playbook.yml
                :target (str (workdir opts) "/ansible-remote/playbook.yml")
                :data {:host (:once/server opts)
                       :smtp (:once/smtp opts)
                       :websites (:once/website opts)
                       :deploy_key (:deploy (:once/ssh opts))}}]]
    (sc/scaffold opts specs)))

;; --- wiring -----------------------------------------------------------------

(defn wire-fn [step _]
  (case step
    :once/start          [start-step :once/compute :once/smtp]
    :once/compute        [compute-step :once/dns]
    :once/smtp           [smtp-step :once/dns]
    :once/dns            [dns-step :once/smtp-post]
    :once/smtp-post      [smtp-post-step :once/ansible-local :once/ansible-remote]
    :once/ansible-local  [ansible-local-step]
    :once/ansible-remote [ansible-remote-step]))

(def workflow
  (-> (wf/workflow {:start :once/start :wire-fn wire-fn})
      ;; provider swap (headline): compute's HCL is chosen by advice.
      ;;   swap to Oracle Cloud by re-adding under the same ::provider id:
      ;;     (wf/advice-add :once/compute :before ::provider oci-provider)
      (wf/advice-add :once/compute :before ::provider digitalocean-provider)
      ;; backend as advice: one isolated tofu state per tofu step. Local here;
      ;; for s3, isolate by a per-step key derived from the same dir, e.g.
      ;;   (tofu/s3-backend-advice server-dir
      ;;     (fn [_] {:bucket "once-tfstate" :key "once/server/terraform.tfstate"
      ;;              :region "eu-west-1"}))
      (wf/advice-add :once/compute   :before ::backend (tofu/local-backend-advice server-dir))
      (wf/advice-add :once/smtp      :before ::backend (tofu/local-backend-advice smtp-dir))
      (wf/advice-add :once/dns       :before ::backend (tofu/local-backend-advice dns-dir))
      (wf/advice-add :once/smtp-post :before ::backend (tofu/local-backend-advice smtp-post-dir))
      ;; ./green create --dry-run prints what would run instead of running it
      (dry-run/advise [:once/compute :once/smtp :once/dns :once/smtp-post
                       :once/ansible-local :once/ansible-remote])))

(defn file-arg? [arg]
  (or (= "-f" arg)
      (= "--file" arg)
      (str/starts-with? (str arg) "--file=")))

(defn default-args [args]
  (let [args (vec args)]
    (cond-> args
      (not-any? file-arg? args)
      (conj "-f" (str (fs/path script-dir "green.edn"))))))

(defn launched-as-script? []
  (when-let [bb-file (System/getProperty "babashka.file")]
    (= (str (fs/canonicalize bb-file))
       (str (fs/canonicalize *file*)))))

(defn run
  "REPL-friendly entrypoint: runs without calling System/exit."
  [& args]
  (cli/run-cli workflow (default-args args)))

(defn -main [& args]
  (cli/exec workflow (default-args args)))

(when (launched-as-script?)
  (apply -main *command-line-args*))

#!/usr/bin/env bb
;; Several fake ONCE boxes from ONE once workflow, composed with wf/step —
;; the multi-zookeeper idiom applied to the once example. Run from here:
;;   ./green create --dry-run   |   ./green create   |   ./green delete
;;
;; Per-deployment graph (one once-wf run per :once/deployments entry):
;;   compute -+                              +-> ansible-local  (~/.ssh/config)
;;            +-> dns -> smtp-post ----------+
;;   smtp ----+                              +-> ansible-remote (playbook)
;;
;; Higher-level graph (the composition layer):
;;   start -> deployment(once-wf) x N -> report
;;
;; Two things the parent swaps on the embedded steps, both the "re-add under
;; the same id" idiom (like multi-zookeeper swapping a child's backend):
;;   ::provider  digitalocean (child default) -> DATA-DRIVEN per deployment
;;               (:eu -> digitalocean, :us -> oci), read from :once/provider.
;;   ::backend   local (child default) -> S3, isolated by a per-deployment +
;;               per-step :key, so no two of the 2xN=8 tofu states collide.
;;
;; NOTE: the S3 backend is DEMONSTRATION-ONLY. `./green create` needs a real
;; bucket + AWS credentials and will not run offline. The offline path is
;; `./green create --dry-run`: the parent's dry-run advice is outermost, so it
;; skips each tofu step AND its :before provider/backend advice — touching
;; nothing (no main.tf, no backend.tf, no S3).
(require '[babashka.fs :as fs])

(def script-dir
  (if (= "NO_SOURCE_PATH" *file*)
    (fs/cwd)
    (fs/parent (fs/canonicalize *file*))))

(def repo-root (fs/canonicalize (fs/path script-dir "../..")))

;; Under babashka, pull in green + this example's resources at runtime.
;; Under the JVM (e.g. cider-jack-in with clojure-cli) these namespaces don't
;; exist and the sibling deps.edn supplies the classpath instead, so the whole
;; block is skipped. requiring-resolve keeps the JVM compiler from ever seeing
;; the babashka-only vars.
(when (System/getProperty "babashka.version")
  ((requiring-resolve 'babashka.deps/add-deps)
   {:deps {'io.github.amiorin/green {:local/root (str repo-root)}}})
  ((requiring-resolve 'babashka.classpath/add-classpath)
   (str (fs/path script-dir "resources"))))

(require '[clojure.java.io :as io]
         '[clojure.string :as str]
         '[green.cli :as cli]
         '[green.dry-run :as dry-run]
         '[green.scaffold :as sc]
         '[green.tofu :as tofu]
         '[green.workflow :as wf])

(defn logln [& xs]
  (locking *out* (apply println xs) (flush)))

;; --- paths: one isolated tofu working dir per step, under work/<deployment> -
;; The parent's :in sets :once/workdir to work/<name>, so server-dir becomes
;; work/eu/server, work/us/server, ... — deployments never share a dir.

(defn example-path [path]
  (when path
    (let [p (fs/path path)]
      (str (if (fs/absolute? p) p (fs/path script-dir p))))))

(defn workdir [opts]
  (example-path (:once/workdir opts)))

(defn step-dir [opts nm] (str (workdir opts) "/" nm))

(defn server-dir [opts] (step-dir opts "server"))

(defn smtp-dir [opts] (step-dir opts "smtp"))

(defn dns-dir [opts] (step-dir opts "dns"))

(defn smtp-post-dir [opts] (step-dir opts "smtp-post"))

;; --- domain discovery from :once/website ------------------------------------

(defn apex
  "example.com from chat.example.com (last two labels)."
  [hostname]
  (->> (str/split hostname #"\.") (take-last 2) (str/join ".")))

(defn sending-domain
  "The smtp identity is always notifications.<domain>.<tld>, derived from the
  website hostnames' apex."
  [opts]
  (str "notifications." (apex (:hostname (first (:once/website opts))))))

(defn keywordize [m]
  (into {} (map (fn [[k v]] [(keyword k) v])) m))

;; --- provider-swap advice (headline), now data-driven per deployment --------
;; Same tofu.clj backend-as-advice idiom: a :before advice that scaffolds the
;; provider-specific main.tf before `compute` runs. In multi-once the choice is
;; per-deployment DATA: the advice reads :once/provider (set by the parent :in
;; from each deployment) and picks the matching mock. The advice is uniform
;; across branches; the data differs — downstream steps still never learn which
;; cloud produced the host.

(def provider-templates
  {:digitalocean :once.do/main.tf
   :oci          :once.oci/main.tf})

(defn provider-advice [dir-fn]
  (fn [opts]
    (let [tmpl (provider-templates (:once/provider opts) :once.do/main.tf)
          dir  (io/file (dir-fn opts))]
      (.mkdirs dir)
      (spit (io/file dir "main.tf")
            (sc/render-template tmpl {:host (:once/host opts)
                                      :ssh  (:once/ssh opts)})))
    opts))

;; --- S3 backend config (demonstration-only) ---------------------------------
;; The bucket is shared, so isolation MUST come from a distinct :key. We derive
;; it from BOTH the deployment name (:once/deployment) and the step's dir
;; segment, e.g. once/eu/server/…, once/us/smtp/… — 2 deployments x 4 tofu
;; steps = 8 separate state objects, none clobbering another. Swap to gcs by
;; re-adding ::backend with gcs-backend-advice and a per-step :prefix.

(defn s3-config [dir-fn]
  (fn [opts]
    {:bucket "once-tfstate"                       ; placeholder — needs a real bucket
     :key    (str "once/" (:once/deployment opts) "/"
                  (last (str/split (dir-fn opts) #"/"))
                  "/terraform.tfstate")
     :region "eu-west-1"}))

;; --- tofu steps -------------------------------------------------------------

(defn start-step [opts] opts)

;; compute is a bare tofu run: its main.tf is written by the ::provider advice
;; and its backend.tf by the ::backend advice, both :before.
(defn compute-step [opts]
  (tofu/tofu-step opts {:dir (server-dir opts)}))

(defn tofu-with-spec
  "Scaffold main.tf then apply (create); or destroy then remove main.tf
  (delete). The zookeeper node-step pattern."
  [opts dir specs]
  (if (= :delete (:green/event opts))
    (let [opts (tofu/tofu-step opts {:dir dir})]
      (if (pos? (:green/exit opts 0)) opts (sc/scaffold opts specs)))
    (let [opts (sc/scaffold opts specs)]
      (tofu/tofu-step opts {:dir dir}))))

(defn smtp-step [opts]
  (let [dir (smtp-dir opts)
        sub (sending-domain opts)
        specs [{:template :once.smtp/main.tf
                :target (str dir "/main.tf")
                :data {:subdomain sub
                       :id (str "snd-" (str/replace (apex (:hostname (first (:once/website opts)))) "." "-"))
                       :username "resend"
                       :password "mock-smtp-secret"}}]]
    (tofu-with-spec opts dir specs)))

;; dns joins compute and smtp: it needs compute.:ip and smtp.:records. It also
;; carries the two provisioning outputs forward (:once/server, :once/smtp) so
;; smtp-post and the ansible steps can read them without :green/branches.
(defn dns-step [opts]
  (let [dir (dns-dir opts)
        outs (map :tofu/outputs (:green/branches opts))
        compute-out (some #(when (:ip %) %) outs)
        smtp-out (some #(when (:records %) %) outs)
        specs [{:template :once.cloudflare/main.tf
                :target (str dir "/main.tf")
                :data {:ip (:ip compute-out)
                       :hostnames (map :hostname (:once/website opts))
                       :email (map keywordize (:records smtp-out))}}]
        opts (assoc opts :once/server compute-out :once/smtp smtp-out)]
    (tofu-with-spec opts dir specs)))

(defn smtp-post-step [opts]
  (let [dir (smtp-post-dir opts)
        specs [{:template :once.smtp-post/main.tf
                :target (str dir "/main.tf")
                :data {:id (:id (:once/smtp opts))
                       :subdomain (sending-domain opts)}}]]
    (tofu-with-spec opts dir specs)))

;; --- scaffold-only steps (no command) ---------------------------------------

;; ansible-local configures the OPERATOR's machine: a ~/.ssh/config Host entry
;; for the new VPS, rendered as a fragment to Include.
(defn ansible-local-step [opts]
  (let [srv (:once/server opts)
        specs [{:template :once/ssh-config
                :target (str (workdir opts) "/ansible-local/config")
                :data {:name (:name srv)
                       :ip (:ip srv)
                       :sudoer (:sudoer srv)
                       :identity "~/.ssh/once_compute"}}]]
    (sc/scaffold opts specs)))

;; ansible-remote configures the BOX: the playbook (Docker + ONCE + sites).
(defn ansible-remote-step [opts]
  (let [specs [{:template :once/playbook.yml
                :target (str (workdir opts) "/ansible-remote/playbook.yml")
                :data {:host (:once/server opts)
                       :smtp (:once/smtp opts)
                       :websites (:once/website opts)
                       :deploy_key (:deploy (:once/ssh opts))}}]]
    (sc/scaffold opts specs)))

;; --- the single-VPS once workflow (same shape as examples/once) -------------
;; Ships digitalocean provider + local backend as its defaults, so it runs
;; standalone. The composition layer below re-adds ::provider and ::backend to
;; swap both — exactly how multi-zookeeper's parent can replace cluster-wf's
;; default backend.

(defn once-wire-fn [step _]
  (case step
    :once/start          [start-step :once/compute :once/smtp]
    :once/compute        [compute-step :once/dns]
    :once/smtp           [smtp-step :once/dns]
    :once/dns            [dns-step :once/smtp-post]
    :once/smtp-post      [smtp-post-step :once/ansible-local :once/ansible-remote]
    :once/ansible-local  [ansible-local-step]
    :once/ansible-remote [ansible-remote-step]))

(def once-wf
  (-> (wf/workflow {:start :once/start :wire-fn once-wire-fn})
      (wf/advice-add :once/compute :before ::provider (provider-advice server-dir))
      (wf/advice-add :once/compute   :before ::backend (tofu/local-backend-advice server-dir))
      (wf/advice-add :once/smtp      :before ::backend (tofu/local-backend-advice smtp-dir))
      (wf/advice-add :once/dns       :before ::backend (tofu/local-backend-advice dns-dir))
      (wf/advice-add :once/smtp-post :before ::backend (tofu/local-backend-advice smtp-post-dir))))

;; --- the composition layer: one once-wf run per deployment ------------------

(defn deployments-wire-fn [step _]
  (case step
    :multi/start [start-step :multi/deployment]
    :multi/deployment
    [(wf/step once-wf
              {:in (fn [opts]
                     (let [d (:multi/deployment opts)]
                       ;; carry ambient keys by assoc'ing onto opts (keeps
                       ;; :green/event, :green/dry-run, inherited advice), then
                       ;; project this deployment's fields into the once keys
                       (assoc opts
                              :once/deployment (:name d)      ; -> S3 key + report
                              :once/provider   (:provider d)  ; -> ::provider advice
                              :once/host       (:host d)
                              :once/ssh        (:ssh d)
                              :once/website    (:website d)
                              :once/workdir    (str (:once/workdir opts) "/" (:name d)))))})
     :multi/report]
    :multi/report
    [(fn [opts]
       (doseq [b (:green/branches opts)]
         (logln "deployment ready:" (:once/deployment b)
                "->" (:once/workdir b)
                (str "(" (name (:once/provider b)) ")")))
       opts)]))

(defn deployments-next-fn [step default-next opts]
  (cond
    (pos? (:green/exit opts 0)) []
    (= step :multi/start) (mapv (fn [d] [:multi/deployment (assoc opts :multi/deployment d)])
                                (:once/deployments opts))
    :else (mapv (fn [s] [s opts]) default-next)))

(def workflow
  (-> (wf/workflow {:start :multi/start
                    :wire-fn deployments-wire-fn
                    :next-fn deployments-next-fn})
      ;; provider swap (headline): replace the child's digitalocean default with
      ;; a data-driven pick, so :eu -> digitalocean and :us -> oci in one run.
      ;; Same ::provider id -> replaces the inherited child advice.
      (wf/advice-add :once/compute :before ::provider (provider-advice server-dir))
      ;; backend swap: replace the child's local backend with S3, isolated by a
      ;; per-deployment + per-step :key. Same ::backend id -> replaces each
      ;; child advice; whichever backend is active, the key keeps all 8 states
      ;; apart. (Demonstration-only: real `create` needs the bucket to exist.)
      (wf/advice-add :once/compute   :before ::backend (tofu/s3-backend-advice server-dir    (s3-config server-dir)))
      (wf/advice-add :once/smtp      :before ::backend (tofu/s3-backend-advice smtp-dir      (s3-config smtp-dir)))
      (wf/advice-add :once/dns       :before ::backend (tofu/s3-backend-advice dns-dir       (s3-config dns-dir)))
      (wf/advice-add :once/smtp-post :before ::backend (tofu/s3-backend-advice smtp-post-dir (s3-config smtp-post-dir)))
      ;; dry-run declared LAST at the parent, so ::skip gets the largest seq and
      ;; stays OUTERMOST over the re-added :before provider/backend advice —
      ;; `./green create --dry-run` then skips them too and touches nothing.
      ;; Inspect the composed stack with e.g.
      ;;   (wf/advice-plan [workflow once-wf] :once/compute)
      (dry-run/advise [:once/compute :once/smtp :once/dns :once/smtp-post
                       :once/ansible-local :once/ansible-remote])))

(defn default-args [args]
  (let [args (vec args)]
    (cond-> args
      (not-any? #(or (= "-f" %)
                     (= "--file" %)
                     (str/starts-with? % "--file="))
                args)
      (conj "-f" (str (fs/path script-dir "green.edn"))))))

(defn launched-as-script? []
  (when-let [bb-file (System/getProperty "babashka.file")]
    (= (str (fs/canonicalize bb-file))
       (str (fs/canonicalize *file*)))))

(defn run
  "REPL-friendly entrypoint: runs without calling System/exit."
  [& args]
  (cli/run-cli workflow (default-args args)))

(defn -main [& args]
  (cli/exec workflow (default-args args)))

(when (launched-as-script?)
  (apply -main *command-line-args*))

#!/usr/bin/env bb
;; Self-contained green CLI for a REAL 3-node ZooKeeper cluster:
;; OpenTofu (AWS provider pointed at floci, a local AWS emulator) creates
;; one Docker-backed EC2 instance per node, and Ansible over SSH provisions
;; ZooKeeper — no user-data provisioning. Linux only: the health check and
;; Ansible talk straight to the instances' Docker-bridge IPs.
;; Run from this directory:   ./green create   |   ./green delete
(require '[babashka.fs :as fs])

(def script-dir
  (if (= "NO_SOURCE_PATH" *file*)
    (fs/cwd)
    (fs/parent (fs/canonicalize *file*))))

(def repo-root (fs/canonicalize (fs/path script-dir "../..")))

;; Under babashka, pull in green + this example's resources at runtime.
;; Under the JVM (e.g. cider-jack-in with clojure-cli) these namespaces don't
;; exist and the sibling deps.edn supplies the classpath instead, so the whole
;; block is skipped. requiring-resolve keeps the JVM compiler from ever seeing
;; the babashka-only vars.
(when (System/getProperty "babashka.version")
  ((requiring-resolve 'babashka.deps/add-deps)
   {:deps {'io.github.amiorin/green {:local/root (str repo-root)}}})
  ((requiring-resolve 'babashka.classpath/add-classpath)
   (str (fs/path script-dir "resources"))))

(require '[cheshire.core :as json]
         '[clojure.java.io :as io]
         '[clojure.java.shell :as shell]
         '[clojure.string :as str]
         '[green.ansible :as ansible]
         '[green.cli :as cli]
         '[green.dry-run :as dry-run]
         '[green.progress :as progress]
         '[green.scaffold :as sc]
         '[green.tofu :as tofu]
         '[green.workflow :as wf])

(def floci-endpoint "http://localhost:4566")

(defn example-path [path]
  (when path
    (let [p (fs/path path)]
      (str (if (fs/absolute? p) p (fs/path script-dir p))))))

(defn workdir [opts]
  (example-path (:zk/workdir opts)))

(defn shared-dir [opts]
  (str (workdir opts) "/shared"))

(defn node-dir [opts node]
  (str (workdir opts) "/nodes/" (:id node)))

(defn ssh-key [opts]
  (str (workdir opts) "/ssh/id_ed25519"))

(defn inventory-file [opts]
  (str (workdir opts) "/ansible/inventory.ini"))

(defn- logln [& xs]
  (locking *out*
    (apply println xs)
    (flush)))

(defn- sh! [& args]
  (let [{:keys [exit err out]} (apply shell/sh args)]
    (when (pos? exit)
      (throw (ex-info (str (first args) " failed: "
                           (or (not-empty err) (not-empty out) "(no output)"))
                      {})))
    out))

;; --- validation advices ------------------------------------------------------
;; :before-while gates: return true to continue; throw ex-info for a clear
;; failure — the engine converts it to :green/exit + :green/err.

(defn check-schema
  "Gate on :zk/start — validate the desired-state shape before anything
  runs. Pure, so it also runs under --dry-run."
  [opts]
  (let [servers (:zk/servers opts)
        problems
        (cond-> []
          (not (string? (:zk/workdir opts))) (conj ":zk/workdir must be a string")
          (not (string? (:zk/ami opts))) (conj ":zk/ami must be a string")
          (not (string? (:zk/user opts))) (conj ":zk/user must be a string")
          (not (and (vector? servers) (seq servers)))
          (conj ":zk/servers must be a non-empty vector")
          (and (sequential? servers)
               (not (every? #(and (pos-int? (:id %)) (string? (:name %))) servers)))
          (conj "every server needs a positive :id and a string :name")
          (and (sequential? servers)
               (not= (count servers) (count (set (map :id servers)))))
          (conj "server :id values must be unique")
          (and (sequential? servers)
               (not= (count servers) (count (set (map :name servers)))))
          (conj "server :name values must be unique"))]
    (when (seq problems)
      (throw (ex-info (str "invalid desired state: " (str/join "; " problems))
                      {:green/exit 2}))))
  true)

(defn- tool? [& cmd]
  (try (zero? (:exit (apply shell/sh cmd)))
       (catch Exception _ false)))

(defn check-requirements
  "Gate on :zk/start — the tools the workflow shells out to. Skipped under
  --dry-run, where nothing will run and nothing is required."
  [opts]
  (when-not (:green/dry-run opts)
    (let [missing (cond-> []
                    (not (tool? "tofu" "version")) (conj "tofu")
                    (not (tool? "ansible-playbook" "--version")) (conj "ansible-playbook"))]
      (when (seq missing)
        (throw (ex-info (str "missing requirements: " (str/join ", " missing))
                        {:green/exit 2})))))
  true)

(defn check-node-input
  "Gate on :zk/node — the fan-out must have scoped a server into opts."
  [opts]
  (when-not (and (map? (:zk/node opts)) (:id (:zk/node opts)))
    (throw (ex-info "step :zk/node needs :zk/node {:id .. :name ..} in opts"
                    {:green/exit 2})))
  true)

;; --- prerequisites (:before = setup) ----------------------------------------

(def ^:private aws-env
  (merge (into {} (System/getenv))
         {"AWS_ACCESS_KEY_ID" "test"
          "AWS_SECRET_ACCESS_KEY" "test"
          "AWS_DEFAULT_REGION" "us-east-1"}))

(defn- running-instances? []
  (try
    (let [out (sh! "aws" "--endpoint-url" floci-endpoint
                   "ec2" "describe-instances"
                   "--filters" "Name=instance-state-name,Values=running,pending"
                   "--query" "Reservations[*].Instances[*].InstanceId"
                   "--output" "json"
                   :env aws-env)]
      (seq (json/parse-string out)))
    (catch Exception _ false)))

(defn- clear-tofu-state!
  "Remove all terraform.tfstate files under workdir — after a floci restart
  they reference resources that no longer exist."
  [opts]
  (doseq [f (file-seq (io/file (workdir opts)))
          :when (and (.isFile f)
                     (str/ends-with? (.getName f) ".tfstate"))]
    (io/delete-file f)))

(defn- floci-healthy? []
  (try (slurp (str floci-endpoint "/_localstack/health")) true
       (catch Exception _ false)))

(defn- wait-floci! []
  (loop [n 60]
    (cond
      (floci-healthy?) true
      (zero? n) (throw (ex-info "floci did not become healthy" {}))
      :else (do (Thread/sleep 1000) (recur (dec n))))))

(defn- remove-ec2-containers! []
  (let [ids (not-empty
              (str/trim
                (sh! "docker" "ps" "-aq" "--filter" "name=floci-ec2")))]
    (when ids
      (apply sh! "docker" "rm" "-f" (str/split-lines ids)))))

(defn- restart-floci! []
  (remove-ec2-containers!)
  (sh! "docker" "restart" "floci")
  (wait-floci!))

(defn ensure-floci
  ":before advice on :zk/start — on create: if no running instances,
  remove stale EC2 containers, restart floci (clears stale key pairs),
  and wipe tofu state; otherwise floci is already running with live
  instances, nothing to do."
  [opts]
  (when-not (:green/dry-run opts)
    (if (running-instances?)
      (logln "floci: instances present — skipping restart")
      (do (logln "floci: no instances — restarting for clean state")
          (restart-floci!)
          (clear-tofu-state! opts))))
  opts)

(defn ensure-ssh-key
  ":before advice on :zk/start — generate the keypair tofu injects and
  Ansible authenticates with. Kept across delete so create/delete cycles
  reuse it; remove the work dir to rotate."
  [opts]
  (when-not (:green/dry-run opts)
    (let [key (ssh-key opts)]
      (when-not (fs/exists? key)
        (fs/create-dirs (fs/parent (fs/path key)))
        (sh! "ssh-keygen" "-t" "ed25519" "-f" key "-N" "" "-q"))))
  opts)

;; --- steps -------------------------------------------------------------------

(defn start-step
  "Create/destroy the shared key pair (one tofu root, run once before the
  per-node fan-out). Skipped under --dry-run."
  [opts]
  (if (:green/dry-run opts)
    opts
    (let [dir (shared-dir opts)
          specs [{:template :zk/shared.tf
                  :target (str dir "/shared.tf")
                  :data {:endpoint floci-endpoint
                         :pubkey (str (ssh-key opts) ".pub")}}]]
      (if (= :delete (:green/event opts))
        (let [opts (tofu/tofu-step opts {:dir dir :output-key :zk/shared})]
          (if (pos? (:green/exit opts 0)) opts (sc/scaffold opts specs)))
        (let [opts (sc/scaffold opts specs)]
          (tofu/tofu-step opts {:dir dir :output-key :zk/shared}))))))

(defn node-step
  "Scaffold one node's main.tf and drive tofu over it. On delete, destroy
  before removing the files tofu still needs."
  [opts]
  (let [node (:zk/node opts)
        dir (node-dir opts node)
        specs [{:template :zk/main.tf
                :target (str dir "/main.tf")
                :data {:node node
                       :ami (:zk/ami opts)
                       :instance-type (:zk/instance-type opts "t3.micro")
                       :endpoint floci-endpoint
                       :key-name "zk"}}]]
    (if (= :delete (:green/event opts))
      (let [opts (tofu/tofu-step opts {:dir dir})]
        (if (pos? (:green/exit opts 0)) opts (sc/scaffold opts specs)))
      (let [opts (sc/scaffold opts specs)]
        (tofu/tofu-step opts {:dir dir})))))

(defn with-observed-nodes
  ":filter-args advice on :zk/ansible — normalize the step's input: the
  observed node list ({:id :name :ip :instance_id}) under :zk/nodes. On
  create it comes from the joined branches' tofu outputs; on delete the
  instances still exist, so it is read back from each node's tofu state
  (nodes with no state are skipped — delete stays idempotent)."
  [opts]
  (assoc opts :zk/nodes
         (if (= :delete (:green/event opts))
           (vec (sort-by :id
                         (keep (fn [n]
                                 (try (let [o (tofu/outputs (node-dir opts n))]
                                        (when (:ip o) o))
                                      (catch Exception _ nil)))
                               (:zk/servers opts))))
           (->> (:green/branches opts) (map :tofu/outputs) (sort-by :id) vec))))

(defn- ssh-open? [ip]
  (try (with-open [sock (java.net.Socket.)]
         (.connect sock (java.net.InetSocketAddress. ^String ip (int 22)) 2000))
       true
       (catch Exception _ false)))

(defn wait-for-ssh
  ":before advice on :zk/ansible — floci reports instances running before
  the sshd inside them accepts connections; wait for every node's :22
  before the playbook needs it. Skipped on delete: the instances have been
  up for a while, and a stale state entry must not stall the teardown."
  [opts]
  (when-not (= :delete (:green/event opts))
    (doseq [{:keys [name ip]} (:zk/nodes opts)]
      (loop [n 60]
        (cond
          (ssh-open? ip) nil
          (zero? n) (throw (ex-info (str "ssh on " name " (" ip ":22) never came up")
                                    {}))
          :else (do (Thread/sleep 3000) (recur (dec n)))))))
  opts)

(defn inventory-data
  "Inventory for green.ansible/inventory-advice, from the observed nodes."
  [opts]
  {"zookeeper"
   {:hosts (mapv (fn [{:keys [id name ip]}]
                   {:name name :vars {:ansible_host ip :zk_id id}})
                 (:zk/nodes opts))
    :vars {:ansible_user (:zk/user opts)
           :ansible_python_interpreter "/usr/bin/python3"}}})

(defn ansible-step
  "Join step: one ansible-playbook run over every observed node — create.yml
  provisions ZooKeeper, delete.yml deprovisions it (see green.ansible)."
  [opts]
  (logln (str "ansible: " (ansible/playbook opts) " over "
              (count (:zk/nodes opts)) " node(s)"
              (when-not (= :delete (:green/event opts))
                " — a first create takes a few minutes")))
  (ansible/ansible-step opts
                        {:dir (str script-dir "/ansible")
                         :inventory (inventory-file opts)
                         :private-key (ssh-key opts)
                         :host-key-checking false}))

(defn- four-letter [ip port word]
  (with-open [sock (java.net.Socket.)]
    (.connect sock (java.net.InetSocketAddress. ^String ip (int port)) 3000)
    (.setSoTimeout sock 3000)
    (let [out (.getOutputStream sock)]
      (.write out (.getBytes ^String word))
      (.flush out)
      (slurp (.getInputStream sock)))))

(defn- server-mode [ip]
  (second (re-find #"Mode: (\w+)" (four-letter ip 2181 "srvr"))))

(defn health-step
  "Ask every node `srvr` on its client port and require exactly one leader
  with the rest followers — real verification that the ensemble formed."
  [opts]
  (let [nodes (:zk/nodes opts)
        modes (mapv (fn [{:keys [name ip]}]
                      {:name name
                       :mode (try (server-mode ip)
                                  (catch Exception e (str "error: " (ex-message e))))})
                    nodes)
        tally (frequencies (map :mode modes))]
    (if (and (seq nodes)
             (= 1 (get tally "leader" 0))
             (= (dec (count nodes)) (get tally "follower" 0)))
      (do (logln "health:" (str/join ", " (map #(str (:name %) "=" (:mode %)) modes)))
          (assoc opts :green/exit 0 :zk/health modes))
      (assoc opts
             :green/exit 1
             :zk/health modes
             :green/err (str "quorum not healthy: " (pr-str modes))))))

(defn retry-advice
  ":around — re-run a converging step until it succeeds or attempts run
  out. Leader election takes a few seconds after first start."
  [attempts pause-ms]
  (fn [f opts]
    (loop [n 1]
      (let [res (f opts)]
        (if (or (zero? (:green/exit res 0)) (>= n attempts))
          res
          (do (logln (str "health: not converged (attempt " n "/" attempts
                          "), retrying in " pause-ms "ms"))
              (Thread/sleep pause-ms)
              (recur (inc n))))))))

;; --- workflow ----------------------------------------------------------------

(defn wire-fn [step run-opts]
  (let [delete? (= :delete (:green/event run-opts))]
    (case step
      :zk/start   [start-step (if delete? :zk/ansible :zk/node)]
      :zk/node    (cond-> [node-step]
                    (not delete?) (conj :zk/ansible))
      :zk/ansible [ansible-step (if delete? :zk/node :zk/health)]
      :zk/health  [health-step])))

(defn next-fn [step default-next opts]
  (let [delete? (= :delete (:green/event opts))]
    (cond
      (pos? (:green/exit opts 0)) []
      ;; create: one tofu apply per server, in parallel; they join at ansible
      (and (= step :zk/start) (not delete?))
      (mapv (fn [n] [:zk/node (assoc opts :zk/node n)]) (:zk/servers opts))
      ;; delete graph is :zk/start -> :zk/ansible -> :zk/node; after the
      ;; deprovisioning playbook, fan out the per-node destroys
      (and (= step :zk/ansible) delete?)
      (mapv (fn [n] [:zk/node (assoc opts :zk/node n)]) (:zk/servers opts))
      :else (mapv (fn [s] [s opts]) default-next))))

(def workflow
  (-> (wf/workflow {:start :zk/start :wire-fn wire-fn :next-fn next-fn})
      ;; shared key pair backend — innermost :before, runs right before start-step
      (wf/advice-add :zk/start :before ::shared-backend
                     (tofu/local-backend-advice shared-dir))
      ;; prerequisites first (:before = setup), gates after: later adds are
      ;; outermost, so validation runs before floci/ssh-key setup fires
      (wf/advice-add :zk/start :before ::floci ensure-floci)
      (wf/advice-add :zk/start :before ::ssh-key ensure-ssh-key)
      (wf/advice-add :zk/start :before-while ::requirements check-requirements)
      (wf/advice-add :zk/start :before-while ::schema check-schema)
      ;; per-node tofu state; the backend is advice, not hardwiring
      (wf/advice-add :zk/node :before ::backend
                     (tofu/local-backend-advice #(node-dir % (:zk/node %))))
      (wf/advice-add :zk/node :before-while ::inputs check-node-input)
      ;; ansible input: ssh wait innermost, then the inventory :before
      ;; advice, then the :filter-args node-list normalizer outermost —
      ;; so both inner advices already see :zk/nodes
      (wf/advice-add :zk/ansible :before ::wait-ssh wait-for-ssh)
      (wf/advice-add :zk/ansible :before ::inventory
                     (ansible/inventory-advice inventory-file inventory-data))
      (wf/advice-add :zk/ansible :filter-args ::nodes with-observed-nodes)
      ;; quorum takes a few seconds to elect; poll instead of failing fast
      (wf/advice-add :zk/health :around ::retry (retry-advice 12 5000))
      ;; progress + dry-run declared last; dry-run outermost so --dry-run
      ;; suppresses both the step and its progress line
      (progress/advise)
      (dry-run/advise [:zk/node :zk/ansible :zk/health])))

;; --- CLI ---------------------------------------------------------------------

(defn file-arg? [arg]
  (or (= "-f" arg)
      (= "--file" arg)
      (.startsWith (str arg) "--file=")))

(defn default-args [args]
  (let [args (vec args)]
    (cond-> args
      (not-any? file-arg? args)
      (conj "-f" (str (fs/path script-dir "green.edn"))))))

(defn launched-as-script? []
  (when-let [bb-file (System/getProperty "babashka.file")]
    (= (str (fs/canonicalize bb-file))
       (str (fs/canonicalize *file*)))))

(defn run
  "REPL-friendly entrypoint: runs without calling System/exit."
  [& args]
  (cli/run-cli workflow (default-args args)))

(defn -main [& args]
  (cli/exec workflow (default-args args)))

(when (launched-as-script?)
  (apply -main *command-line-args*))