diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,18 @@ # Changelog +## 0.3.0.0 — 2026-10-28 + +### Breaking Changes + +- Add `KirokuCompaction` and `KirokuStore` constructors to `KirokuCommand`; exhaustive host matches need new arms. + +### New Features + +- Require `kiroku-store ^>=0.13.0.0`; maintenance outcomes use exit 0 for success/already applied, 2 for refusal and 1 for file/usage/store errors. + +- Embed manifest preview, digest-confirmed apply, ledger listing and store identity commands in a host owning a `KirokuStore`. The standalone remote client refuses these commands with guidance. +- Return structured exit codes with `KirokuCommandOutcome` and `executeKirokuCommandWithStore`; expose validated manifest loading and table/JSON renderers in `Kiroku.Cli.Compaction`. + ## 0.2.0.11 — 2026-10-26 ### Other Changes diff --git a/kiroku-cli.cabal b/kiroku-cli.cabal --- a/kiroku-cli.cabal +++ b/kiroku-cli.cabal @@ -1,6 +1,6 @@ cabal-version: 3.0 name: kiroku-cli -version: 0.2.0.11 +version: 0.3.0.0 synopsis: Embeddable operator CLI for Kiroku description: Parser, runner, and executable entry points for Kiroku operator commands. @@ -36,6 +36,7 @@ exposed-modules: Kiroku.Cli Kiroku.Cli.Command + Kiroku.Cli.Compaction Kiroku.Cli.Parser Kiroku.Cli.Run Kiroku.Cli.Standalone @@ -50,10 +51,12 @@ , http-client >=0.7 && <0.8 , http-client-tls >=0.3 && <0.4 , http-types >=0.12 && <0.13 - , kiroku-store ^>=0.11.0.0 || ^>=0.12.0.0 + , kiroku-store ^>=0.13.0.0 , lens >=5.2 && <5.4 , optparse-applicative >=0.19 && <0.20 , text >=2.0 && <2.2 + , uuid >=1.3 && <1.4 + , vector >=0.13 && <0.14 hs-source-dirs: src @@ -72,6 +75,7 @@ import: common type: exitcode-stdio-1.0 main-is: Main.hs + other-modules: Test.Compaction hs-source-dirs: test ghc-options: -threaded -rtsopts -with-rtsopts=-N build-depends: @@ -83,8 +87,11 @@ , http-client >=0.7 && <0.8 , http-client-tls >=0.3 && <0.4 , kiroku-cli - , kiroku-store ^>=0.11.0.0 || ^>=0.12.0.0 + , kiroku-store ^>=0.13.0.0 , kiroku-test-support , lens , optparse-applicative + , temporary >=1.3 && <1.4 , text + , uuid >=1.3 && <1.4 + , vector >=0.13 && <0.14 diff --git a/src/Kiroku/Cli.hs b/src/Kiroku/Cli.hs --- a/src/Kiroku/Cli.hs +++ b/src/Kiroku/Cli.hs @@ -6,9 +6,11 @@ runKirokuCommand, runKirokuCommandWithStore, renderKirokuCommandWithStore, + KirokuCommandOutcome (..), + executeKirokuCommandWithStore, ) where import Kiroku.Cli.Command (KirokuCommand (..)) import Kiroku.Cli.Parser (kirokuCommandParser, kirokuParserInfo, kirokuSubparser) -import Kiroku.Cli.Run (renderKirokuCommandWithStore, runKirokuCommand, runKirokuCommandWithStore) +import Kiroku.Cli.Run (KirokuCommandOutcome (..), executeKirokuCommandWithStore, renderKirokuCommandWithStore, runKirokuCommand, runKirokuCommandWithStore) diff --git a/src/Kiroku/Cli/Command.hs b/src/Kiroku/Cli/Command.hs --- a/src/Kiroku/Cli/Command.hs +++ b/src/Kiroku/Cli/Command.hs @@ -4,14 +4,24 @@ RemoteEndpoint (..), StatusOptions (..), SubscriptionCommand (..), + CompactionCommand (..), + StoreCommand (..), + ManifestPath (..), + CompactionOptions (..), + ApplyOptions (..), + LedgerOptions (..), + IdentityOptions (..), ) where +import Data.Int (Int32) import Data.Text (Text) data KirokuCommand = KirokuNoCommand | KirokuSubscriptions SubscriptionCommand + | KirokuCompaction CompactionCommand + | KirokuStore StoreCommand deriving stock (Eq, Show) data SubscriptionCommand @@ -34,4 +44,32 @@ data OutputFormat = OutputTable | OutputJson + deriving stock (Eq, Show) + +-- | Maintenance commands run only in a host that owns a live store. +data CompactionCommand + = CompactionPreview CompactionOptions + | CompactionApply ApplyOptions + | CompactionLedger LedgerOptions + deriving stock (Eq, Show) + +data StoreCommand = ShowStoreIdentity IdentityOptions + deriving stock (Eq, Show) + +newtype ManifestPath = ManifestPath FilePath + deriving stock (Eq, Show) + +data CompactionOptions = CompactionOptions + {manifestPath :: !ManifestPath, outputFormat :: !OutputFormat} + deriving stock (Eq, Show) + +data ApplyOptions = ApplyOptions + {manifestPath :: !ManifestPath, confirmDigest :: !Text, outputFormat :: !OutputFormat} + deriving stock (Eq, Show) + +data LedgerOptions = LedgerOptions + {limit :: !Int32, outputFormat :: !OutputFormat} + deriving stock (Eq, Show) + +newtype IdentityOptions = IdentityOptions {outputFormat :: OutputFormat} deriving stock (Eq, Show) diff --git a/src/Kiroku/Cli/Compaction.hs b/src/Kiroku/Cli/Compaction.hs new file mode 100644 --- /dev/null +++ b/src/Kiroku/Cli/Compaction.hs @@ -0,0 +1,180 @@ +-- | Pure renderers and file loading shared by embeddable operator hosts. +module Kiroku.Cli.Compaction + ( loadCompactionManifest, + CompactionCommandError (..), + renderCompactionReport, + renderCompactionRefusals, + renderCompactionRecords, + renderStoreIdentity, + renderCompactionCommandError, + renderCompactionResult, + ) +where + +import Control.Exception (IOException, displayException, try) +import Control.Lens ((^.)) +import Data.Aeson qualified as Aeson +import Data.Aeson.KeyMap qualified as KeyMap +import Data.ByteString qualified as BS +import Data.ByteString.Lazy qualified as LBS +import Data.Generics.Labels () +import Data.Int (Int32) +import Data.List.NonEmpty (NonEmpty) +import Data.List.NonEmpty qualified as NE +import Data.Text (Text) +import Data.Text qualified as T +import Data.Text.Encoding qualified as TE +import Data.UUID qualified as UUID +import Data.Vector (Vector) +import Data.Vector qualified as V +import Kiroku.Cli.Command (ManifestPath (..), OutputFormat (..)) +import Kiroku.Store.Compaction.Types +import Kiroku.Store.Error (StoreError (..)) +import Kiroku.Store.Types (GlobalPosition (..), StoreIdentity (..), StreamName (..), StreamVersion (..)) + +data CompactionCommandError + = ManifestUnreadable FilePath Text + | ManifestInvalid FilePath Text + | ConfirmDigestUnparseable Text + | ConfirmDigestMismatch {typed :: Text, manifest :: Text} + | LedgerLimitOutOfRange Int32 + | StoreFailure StoreError + deriving stock (Eq, Show) + +loadCompactionManifest :: ManifestPath -> IO (Either CompactionCommandError CompactionManifest) +loadCompactionManifest (ManifestPath path) = do + readResult <- try (BS.readFile path) :: IO (Either IOException BS.ByteString) + pure $ case readResult of + Left err -> Left (ManifestUnreadable path (T.pack (displayException err))) + Right bytes -> case Aeson.eitherDecodeStrict bytes of + Left err -> Left (ManifestInvalid path (T.pack err)) + Right value -> Right value + +json :: (Aeson.ToJSON a) => a -> Text +json = TE.decodeUtf8 . LBS.toStrict . Aeson.encode + +shown :: (Show a) => a -> Text +shown = T.pack . show + +identityText :: StoreIdentity -> Text +identityText (StoreIdentity uid) = UUID.toText uid + +positionText :: GlobalPosition -> Text +positionText (GlobalPosition n) = shown n + +policyText :: (Aeson.ToJSON a) => a -> Text +policyText value = case Aeson.toJSON value of + Aeson.String text -> text + other -> json other + +renderCompactionReport :: OutputFormat -> Maybe CompactionRecord -> CompactionReport -> Text +renderCompactionReport OutputJson Nothing report = json report +renderCompactionReport OutputJson (Just record) _ = json record +renderCompactionReport OutputTable context report = + maybe "" (\record -> "COMPACTION_ID " <> compactionIdText (record ^. #compactionId) <> "\n") context + <> table + ["FIELD", "VALUE"] + [ ["MANIFEST_DIGEST", compactionDigestHex (report ^. #manifestDigest)], + ["STORE_IDENTITY", identityText (report ^. #storeIdentity)], + ["OPERATION", compactionOperationText (report ^. #operation)], + ["DEAD_LETTER_POLICY", policyText (report ^. #deadLetterPolicy)], + ["CAUSATION_POLICY", policyText (report ^. #causationPolicy)], + ["SELECTED_EVENTS", shown (report ^. #selectedEvents)], + ["HOME_MEMBERSHIPS", shown (report ^. #homeMemberships)], + ["GLOBAL_MEMBERSHIPS", shown (report ^. #globalMemberships)], + ["LINK_MEMBERSHIPS", shown (report ^. #linkMemberships)], + ["DEAD_LETTERS_REMOVED", shown (report ^. #deadLettersRemoved)], + ["CAUSATION_DEPENDENTS", shown (report ^. #causationDependents)], + ["LOWEST_GLOBAL_POSITION", positionText (report ^. #lowestGlobalPosition)], + ["HIGHEST_GLOBAL_POSITION", positionText (report ^. #highestGlobalPosition)], + ["REPORT_DIGEST", compactionDigestHex (report ^. #reportDigest)] + ] + <> "\nAFFECTED_STREAMS\n" + <> table + ["STREAM", "HEAD_VERSION"] + [[name, shown version] | StreamHeadWitness (StreamName name) (StreamVersion version) <- V.toList (report ^. #affectedStreams)] + +renderCompactionRefusals :: OutputFormat -> NonEmpty CompactionRefusal -> Text +renderCompactionRefusals OutputJson refusals = json (NE.toList refusals) +renderCompactionRefusals OutputTable refusals = + "refused: " + <> shown (NE.length refusals) + <> " refusal(s); no rows were changed\n" + <> table ["KIND", "EVENT_ID", "STREAM", "DETAIL"] (map row (NE.toList refusals)) + where + row refusal = case Aeson.toJSON refusal of + Aeson.Object object -> [tag refusal, optional "event_id" object, optional "stream" object, shown refusal] + _ -> [tag refusal, "", "", shown refusal] + optional key object = case KeyMap.lookup key object of + Just (Aeson.String text) -> text + _ -> "" + tag = refusalName + +refusalName :: CompactionRefusal -> Text +refusalName = \case + CompactionStoreIdentityMismatch {} -> "CompactionStoreIdentityMismatch" + CompactionHistoryRetentionActive {} -> "CompactionHistoryRetentionActive" + CompactionStreamMissing {} -> "CompactionStreamMissing" + CompactionStreamSoftDeleted {} -> "CompactionStreamSoftDeleted" + CompactionStreamHeadDrift {} -> "CompactionStreamHeadDrift" + CompactionSelectedEventMissing {} -> "CompactionSelectedEventMissing" + CompactionWitnessMismatch {} -> "CompactionWitnessMismatch" + CompactionUnexpectedLink {} -> "CompactionUnexpectedLink" + CompactionAcknowledgedLinkMissing {} -> "CompactionAcknowledgedLinkMissing" + CompactionDeadLettersPresent {} -> "CompactionDeadLettersPresent" + CompactionCausationDependentsPresent {} -> "CompactionCausationDependentsPresent" + CompactionLedgerConflict {} -> "CompactionLedgerConflict" + +renderCompactionRecords :: OutputFormat -> Vector CompactionRecord -> Text +renderCompactionRecords OutputJson = json +renderCompactionRecords OutputTable = table ["COMPACTION_ID", "APPLIED_AT", "APPLIED_BY", "OPERATION", "SELECTED_EVENTS", "MANIFEST_DIGEST"] . map row . V.toList + where + row record = + [ compactionIdText (record ^. #compactionId), + shown (record ^. #appliedAt), + record ^. #appliedBy, + compactionOperationText (record ^. #report . #operation), + shown (record ^. #report . #selectedEvents), + compactionDigestHex (record ^. #report . #manifestDigest) + ] + +compactionIdText :: CompactionId -> Text +compactionIdText (CompactionId uid) = UUID.toText uid + +renderStoreIdentity :: OutputFormat -> StoreIdentity -> Text +renderStoreIdentity OutputTable identity = table ["STORE_IDENTITY"] [[identityText identity]] +renderStoreIdentity OutputJson identity = json (Aeson.object ["store_identity" Aeson..= identity]) + +-- | JSON remains a single valid document; human status prefixes are table-only. +renderCompactionResult :: OutputFormat -> Bool -> CompactionRecord -> Text +renderCompactionResult OutputJson already record = json (Aeson.object ["result" Aeson..= (if already then "already_applied" else "applied" :: Text), "record" Aeson..= record]) +renderCompactionResult OutputTable already record = + (if already then "already applied: " else "applied: ") + <> compactionIdText (record ^. #compactionId) + <> " at " + <> shown (record ^. #appliedAt) + <> " by " + <> record ^. #appliedBy + <> "\n" + <> renderCompactionReport OutputTable Nothing (record ^. #report) + +renderCompactionCommandError :: OutputFormat -> CompactionCommandError -> Text +renderCompactionCommandError OutputJson err = json (Aeson.object ["error" Aeson..= errorText err]) +renderCompactionCommandError OutputTable err = "kiroku: " <> errorText err + +errorText :: CompactionCommandError -> Text +errorText = \case + ManifestUnreadable path detail -> "cannot read manifest " <> T.pack path <> ": " <> detail + ManifestInvalid path detail -> "invalid manifest " <> T.pack path <> ": " <> detail + ConfirmDigestUnparseable value -> "invalid confirmation digest: " <> value + ConfirmDigestMismatch typed expected -> "confirmation digest " <> typed <> " does not match manifest " <> expected + LedgerLimitOutOfRange value -> "ledger limit must be 1–1000; got " <> shown value + StoreFailure (CompactionConcurrentMutation digest guard) -> + "compaction " <> compactionDigestHex digest <> ": " <> guard <> "; a concurrent reference appeared mid-apply; nothing was deleted — re-run preview and apply" + StoreFailure err -> shown err + +table :: [Text] -> [[Text]] -> Text +table headers rows = T.intercalate "\n" (map render (headers : rows)) + where + widths = [maximum (map T.length (header : [row !! i | row <- rows])) | (i, header) <- zip [0 ..] headers] + render columns = T.stripEnd (T.intercalate " " (zipWith (\width value -> T.justifyLeft width ' ' value) widths columns)) diff --git a/src/Kiroku/Cli/Parser.hs b/src/Kiroku/Cli/Parser.hs --- a/src/Kiroku/Cli/Parser.hs +++ b/src/Kiroku/Cli/Parser.hs @@ -14,6 +14,7 @@ Mod, Parser, ParserInfo, + auto, command, eitherReader, fullDesc, @@ -42,6 +43,8 @@ (KirokuSubscriptions <$> subscriptionCommandParser) (fullDesc <> progDesc "Inspect live subscriptions in this process's KirokuStore registry.") ) + <> command "compaction" (info (KirokuCompaction <$> compactionCommandParser) (fullDesc <> progDesc "Preview, apply, or audit a selective compaction manifest (embeddable only).")) + <> command "store" (info (KirokuStore <$> storeCommandParser) (fullDesc <> progDesc "Inspect the live store identity (embeddable only).")) ) ) @@ -101,3 +104,20 @@ <> progDesc "Run Kiroku operator commands." ) ) + +compactionCommandParser :: Parser Command.CompactionCommand +compactionCommandParser = + subparser + ( command "preview" (info (Command.CompactionPreview <$> (Command.CompactionOptions <$> manifestPathParser <*> outputFormatParser) <**> helper) (progDesc "Validate a manifest without changing any rows.")) + <> command "apply" (info (Command.CompactionApply <$> (Command.ApplyOptions <$> manifestPathParser <*> (T.pack <$> strOption (long "confirm-digest" <> metavar "HEX" <> help "Reviewed manifest digest; must match the file.")) <*> outputFormatParser) <**> helper) (progDesc "Apply an explicitly confirmed manifest atomically.")) + <> command "ledger" (info (Command.CompactionLedger <$> (Command.LedgerOptions <$> option auto (long "limit" <> metavar "N" <> value 50 <> help "Newest records to show (1–1000).") <*> outputFormatParser) <**> helper) (progDesc "List durable compaction audit records.")) + ) + +storeCommandParser :: Parser Command.StoreCommand +storeCommandParser = subparser (command "identity" (info (Command.ShowStoreIdentity . Command.IdentityOptions <$> outputFormatParser <**> helper) (progDesc "Read this store's immutable identity."))) + +manifestPathParser :: Parser Command.ManifestPath +manifestPathParser = Command.ManifestPath <$> strOption (long "manifest" <> metavar "FILE" <> help "Digest-sealed compaction manifest JSON file.") + +outputFormatParser :: Parser Command.OutputFormat +outputFormatParser = option (eitherReader parseOutputFormat) (long "format" <> metavar "table|json" <> value Command.OutputTable <> help "Render a human table or JSON.") diff --git a/src/Kiroku/Cli/Run.hs b/src/Kiroku/Cli/Run.hs --- a/src/Kiroku/Cli/Run.hs +++ b/src/Kiroku/Cli/Run.hs @@ -2,38 +2,91 @@ ( runKirokuCommand, runKirokuCommandWithStore, renderKirokuCommandWithStore, + KirokuCommandOutcome (..), + executeKirokuCommandWithStore, ) where +import Control.Monad (when) import Data.Text (Text) import Data.Text.IO qualified as TIO -import Kiroku.Cli.Command (KirokuCommand (..), StatusOptions (..), SubscriptionCommand (..)) -import Kiroku.Cli.Subscription.Status - ( renderRemoteSubscriptionStatus, - renderSubscriptionStatusRows, - subscriptionStatusRows, - ) +import Kiroku.Cli.Command +import Kiroku.Cli.Compaction +import Kiroku.Cli.Subscription.Status (renderRemoteSubscriptionStatus, renderSubscriptionStatusRows, subscriptionStatusRows) import Kiroku.Store (KirokuStore) +import Kiroku.Store.Compaction qualified as Store +import Kiroku.Store.Effect (runStoreIO) +import Kiroku.Store.Read (storeIdentity) import Kiroku.Store.Subscription (subscriptionStates) +import System.Exit (ExitCode (..), exitWith) +data KirokuCommandOutcome = KirokuCommandOutcome {output :: !Text, exitCode :: !ExitCode} + deriving stock (Eq, Show) + runKirokuCommand :: KirokuCommand -> IO () -runKirokuCommand KirokuNoCommand = - putStrLn "No Kiroku operator command was selected." -runKirokuCommand (KirokuSubscriptions _) = - putStrLn "This command needs a live KirokuStore. Use runKirokuCommandWithStore from the embeddable library API." +runKirokuCommand KirokuNoCommand = putStrLn "No Kiroku operator command was selected." +runKirokuCommand (KirokuSubscriptions _) = putStrLn storeRequired +runKirokuCommand (KirokuCompaction _) = putStrLn storeRequired +runKirokuCommand (KirokuStore _) = putStrLn storeRequired +storeRequired :: String +storeRequired = "This command needs a live KirokuStore. Use runKirokuCommandWithStore from the embeddable library API." + runKirokuCommandWithStore :: KirokuStore -> KirokuCommand -> IO () -runKirokuCommandWithStore store command = - renderKirokuCommandWithStore store command >>= TIO.putStrLn +runKirokuCommandWithStore store command = do + KirokuCommandOutcome {output, exitCode} <- executeKirokuCommandWithStore store command + TIO.putStrLn output + when (exitCode /= ExitSuccess) (exitWith exitCode) renderKirokuCommandWithStore :: KirokuStore -> KirokuCommand -> IO Text -renderKirokuCommandWithStore _ KirokuNoCommand = - pure "No Kiroku operator command was selected." -renderKirokuCommandWithStore store (KirokuSubscriptions (SubscriptionStatus (StatusOptions format endpoint))) = - case endpoint of - -- An optional remote override: a host can point the in-process command at - -- a sibling worker over HTTP instead of reading its own registry. +renderKirokuCommandWithStore store command = do + KirokuCommandOutcome {output} <- executeKirokuCommandWithStore store command + pure output + +-- | Structured execution for hosts. Refusals use exit 2, load/store errors use +-- exit 1, and success/already-applied use exit 0. Legacy status behavior stays +-- compatible, including its rendered remote-fetch errors with exit 0. +executeKirokuCommandWithStore :: KirokuStore -> KirokuCommand -> IO KirokuCommandOutcome +executeKirokuCommandWithStore _ KirokuNoCommand = pure (success "No Kiroku operator command was selected.") +executeKirokuCommandWithStore store (KirokuSubscriptions (SubscriptionStatus (StatusOptions format endpoint))) = + success <$> case endpoint of Just ep -> renderRemoteSubscriptionStatus ep format - Nothing -> do - snapshot <- subscriptionStates store - pure (renderSubscriptionStatusRows format (subscriptionStatusRows snapshot)) + Nothing -> renderSubscriptionStatusRows format . subscriptionStatusRows <$> subscriptionStates store +executeKirokuCommandWithStore store (KirokuStore (ShowStoreIdentity (IdentityOptions format))) = do + result <- runStoreIO store storeIdentity + pure $ either (failure format 1 . StoreFailure) (success . renderStoreIdentity format) result +executeKirokuCommandWithStore store (KirokuCompaction command) = case command of + CompactionPreview (CompactionOptions path format) -> withManifest path format $ \manifest -> do + result <- runStoreIO store (Store.previewCompaction manifest) + pure $ case result of + Left err -> failure format 1 (StoreFailure err) + Right (Left refusals) -> KirokuCommandOutcome (renderCompactionRefusals format refusals) (ExitFailure 2) + Right (Right (Store.CompactionPreviewReady report)) -> success (renderCompactionReport format Nothing report) + Right (Right (Store.CompactionPreviewAlreadyApplied record)) -> success (renderCompactionResult format True record) + CompactionApply (ApplyOptions path typedDigest format) -> withManifest path format $ \manifest -> + case Store.parseCompactionDigestHex typedDigest of + Nothing -> pure (failure format 1 (ConfirmDigestUnparseable typedDigest)) + Just digest + | digest /= Store.compactionManifestDigest manifest -> + pure (failure format 2 (ConfirmDigestMismatch typedDigest (Store.compactionDigestHex (Store.compactionManifestDigest manifest)))) + Just _ -> do + result <- runStoreIO store (Store.applyCompaction manifest) + pure $ case result of + Left err -> failure format 1 (StoreFailure err) + Right (Left refusals) -> KirokuCommandOutcome (renderCompactionRefusals format refusals) (ExitFailure 2) + Right (Right (Store.CompactionAppliedNow record)) -> success (renderCompactionResult format False record) + Right (Right (Store.CompactionAlreadyApplied record)) -> success (renderCompactionResult format True record) + CompactionLedger (LedgerOptions limit format) -> case Store.mkCompactionLedgerLimit limit of + Left _ -> pure (failure format 1 (LedgerLimitOutOfRange limit)) + Right validated -> do + result <- runStoreIO store (Store.compactionLedger (Store.CompactionLedgerQuery validated)) + pure $ either (failure format 1 . StoreFailure) (success . renderCompactionRecords format) result + +withManifest :: ManifestPath -> OutputFormat -> (Store.CompactionManifest -> IO KirokuCommandOutcome) -> IO KirokuCommandOutcome +withManifest path format continuation = loadCompactionManifest path >>= either (pure . failure format 1) continuation + +success :: Text -> KirokuCommandOutcome +success text = KirokuCommandOutcome text ExitSuccess + +failure :: OutputFormat -> Int -> CompactionCommandError -> KirokuCommandOutcome +failure format code err = KirokuCommandOutcome (renderCompactionCommandError format err) (ExitFailure code) diff --git a/src/Kiroku/Cli/Standalone.hs b/src/Kiroku/Cli/Standalone.hs --- a/src/Kiroku/Cli/Standalone.hs +++ b/src/Kiroku/Cli/Standalone.hs @@ -55,7 +55,7 @@ <> progDesc "Inspect a running Kiroku worker over HTTP. Subscription status queries the worker's \ \kiroku-metrics /subscriptions endpoint (--remote-url or KIROKU_REMOTE_URL); the binary \ - \opens no store of its own." + \opens no store of its own. Compaction commands are embeddable-only." <> header "kiroku - remote operator client for Kiroku event stores" ) @@ -71,6 +71,8 @@ resolveStandaloneOptions env (StandaloneOptions cmd) = case cmd of KirokuNoCommand -> Right (StandaloneRuntime KirokuNoCommand) + KirokuCompaction _ -> Left embeddableOnlyMessage + KirokuStore _ -> Left embeddableOnlyMessage KirokuSubscriptions (SubscriptionStatus (StatusOptions format mEndpoint)) -> case mEndpoint <|> envEndpoint of Just ep -> @@ -93,9 +95,14 @@ runStandaloneCommand (StandaloneRuntime cmd) = case cmd of KirokuNoCommand -> pure "No Kiroku operator command was selected." + KirokuCompaction _ -> pure embeddableOnlyMessage + KirokuStore _ -> pure embeddableOnlyMessage KirokuSubscriptions (SubscriptionStatus (StatusOptions format (Just ep))) -> renderRemoteSubscriptionStatus ep format KirokuSubscriptions (SubscriptionStatus (StatusOptions _ Nothing)) -> -- Unreachable: resolveStandaloneOptions fills the endpoint or fails. pure "kiroku: no worker endpoint; pass --remote-url or set KIROKU_REMOTE_URL." + +embeddableOnlyMessage :: Text +embeddableOnlyMessage = "kiroku: compaction and store identity commands run only inside a host that owns a KirokuStore; the standalone binary is a remote client" diff --git a/test/Main.hs b/test/Main.hs --- a/test/Main.hs +++ b/test/Main.hs @@ -29,6 +29,7 @@ (<**>), ) import Options.Applicative qualified as Options +import Test.Compaction qualified as Compaction import Test.Hspec data HostCommand @@ -39,6 +40,7 @@ main :: IO () main = hspec $ do + Compaction.spec describe "kirokuParserInfo" $ do it "renders top-level help" $ do let result = execParserPure defaultPrefs kirokuParserInfo ["--help"] diff --git a/test/Test/Compaction.hs b/test/Test/Compaction.hs new file mode 100644 --- /dev/null +++ b/test/Test/Compaction.hs @@ -0,0 +1,198 @@ +module Test.Compaction (spec) where + +import Control.Concurrent (myThreadId) +import Control.Exception (try) +import Control.Lens ((&), (.~), (^.)) +import Control.Monad (forM_, when) +import Data.Aeson qualified as Aeson +import Data.Aeson.KeyMap qualified as KM +import Data.Either (isRight) +import Data.Generics.Labels () +import Data.IORef (modifyIORef', newIORef, readIORef) +import Data.List.NonEmpty (NonEmpty (..)) +import Data.Text (Text) +import Data.Text qualified as T +import Data.Text.Encoding qualified as TE +import Data.Text.IO qualified as TIO +import Data.UUID qualified as UUID +import Data.Vector qualified as V +import Kiroku.Cli.Command qualified as C +import Kiroku.Cli.Compaction +import Kiroku.Cli.Parser +import Kiroku.Cli.Run +import Kiroku.Cli.Standalone +import Kiroku.Store hiding (id) +import Kiroku.Test.Postgres (withMigratedTestDatabase) +import Options.Applicative qualified as Options +import System.Exit (ExitCode (..)) +import System.IO.Temp (withSystemTempDirectory) +import Test.Hspec + +spec :: Spec +spec = describe "compaction CLI" $ do + it "parses all maintenance commands and requires explicit apply confirmation" $ do + let cases = + [ (["compaction", "preview", "--manifest", "m.json"], C.KirokuCompaction (C.CompactionPreview (C.CompactionOptions (C.ManifestPath "m.json") C.OutputTable))), + (["compaction", "apply", "--manifest", "m.json", "--confirm-digest", "ab12", "--format", "json"], C.KirokuCompaction (C.CompactionApply (C.ApplyOptions (C.ManifestPath "m.json") "ab12" C.OutputJson))), + (["compaction", "ledger"], C.KirokuCompaction (C.CompactionLedger (C.LedgerOptions 50 C.OutputTable))), + (["compaction", "ledger", "--limit", "5"], C.KirokuCompaction (C.CompactionLedger (C.LedgerOptions 5 C.OutputTable))), + (["store", "identity"], C.KirokuStore (C.ShowStoreIdentity (C.IdentityOptions C.OutputTable))) + ] + forM_ cases $ \(args, expected) -> case Options.execParserPure Options.defaultPrefs kirokuParserInfo args of + Options.Success actual -> actual `shouldBe` expected + _ -> expectationFailure ("could not parse " <> show args) + Options.execParserPure Options.defaultPrefs kirokuParserInfo ["compaction", "apply", "--manifest", "m.json"] `shouldSatisfy` \case Options.Failure _ -> True; _ -> False + + it "embeds the same compaction command under an unrelated host parser" $ do + let parser = Options.info (Options.subparser (kirokuSubparser Just)) Options.fullDesc + case Options.execParserPure Options.defaultPrefs parser ["kiroku", "compaction", "preview", "--manifest", "m.json"] of + Options.Success actual -> actual `shouldBe` Just (previewCommand "m.json" C.OutputTable) + _ -> expectationFailure "nested compaction parser failed" + + it "refuses compaction and identity in the standalone remote client" $ do + let commands = [previewCommand "m.json" C.OutputTable, C.KirokuCompaction (C.CompactionLedger (C.LedgerOptions 50 C.OutputTable)), C.KirokuStore (C.ShowStoreIdentity (C.IdentityOptions C.OutputTable))] + forM_ commands $ \command -> case resolveStandaloneOptions [] (StandaloneOptions command) of + Left message -> message `shouldBe` embeddableMessage + Right _ -> expectationFailure "standalone accepted a store-owned maintenance command" + + it "returns typed file errors and detects corrupted manifest digests before store access" $ + withSystemTempDirectory "kiroku-manifest" $ \directory -> do + let path = directory <> "/manifest.json" + missing <- loadCompactionManifest (C.ManifestPath path) + missing `shouldSatisfy` \case Left (ManifestUnreadable _ _) -> True; _ -> False + TIO.writeFile path "not JSON" + invalid <- loadCompactionManifest (C.ManifestPath path) + invalid `shouldSatisfy` \case Left (ManifestInvalid _ _) -> True; _ -> False + let Aeson.Object object = Aeson.toJSON sampleManifest + Aeson.encodeFile path (Aeson.Object (KM.insert "digest" (Aeson.String (T.replicate 64 "0")) object)) + damaged <- loadCompactionManifest (C.ManifestPath path) + damaged `shouldSatisfy` \case Left (ManifestInvalid _ message) -> T.isInfixOf "CompactionManifestDigestMismatch" message; _ -> False + + it "renders stable empty-ledger, identity and refusal tables and valid JSON" $ do + renderCompactionRecords C.OutputTable V.empty `shouldBe` "COMPACTION_ID APPLIED_AT APPLIED_BY OPERATION SELECTED_EVENTS MANIFEST_DIGEST" + renderStoreIdentity C.OutputTable (StoreIdentity UUID.nil) `shouldBe` "STORE_IDENTITY\n00000000-0000-0000-0000-000000000000" + renderStoreIdentity C.OutputJson (StoreIdentity UUID.nil) `shouldBe` "{\"store_identity\":\"00000000-0000-0000-0000-000000000000\"}" + let refusals = CompactionSelectedEventMissing (EventId UUID.nil) :| [] + table = renderCompactionRefusals C.OutputTable refusals + table `shouldSatisfy` T.isPrefixOf "refused: 1 refusal(s); no rows were changed\nKIND" + table `shouldSatisfy` T.isInfixOf "CompactionSelectedEventMissing" + validJSON (renderCompactionRefusals C.OutputJson refusals) + validJSON (renderCompactionRecords C.OutputJson V.empty) + let digest = compactionManifestDigest sampleManifest + renderCompactionCommandError C.OutputTable (StoreFailure (CompactionConcurrentMutation digest "dead-letter count")) `shouldSatisfy` T.isInfixOf "re-run preview and apply" + + it "executes preview, confirmed apply, no-op reapply, ledger and identity against a live store" $ + withSystemTempDirectory "kiroku-live-manifest" $ \directory -> withTestStore $ \store -> do + let path = directory <> "/manifest.json" + driftPath = directory <> "/drift.json" + input <- liveInput store + let Right m = mkCompactionManifest input + Right drift = mkCompactionManifest (input & #streamHeads .~ [StreamHeadWitness (StreamName "orders-cli") (StreamVersion 4)]) + digest = compactionDigestHex (compactionManifestDigest m) + Aeson.encodeFile path m + Aeson.encodeFile driftPath drift + identity <- executeKirokuCommandWithStore store (C.KirokuStore (C.ShowStoreIdentity (C.IdentityOptions C.OutputJson))) + assertOutcome ExitSuccess "store_identity" identity + assertJSON identity + ready <- executeKirokuCommandWithStore store (previewCommand path C.OutputTable) + assertOutcome ExitSuccess "SELECTED_EVENTS" ready + readyJSON <- executeKirokuCommandWithStore store (previewCommand path C.OutputJson) + assertOutcome ExitSuccess "selected_events" readyJSON + assertJSON readyJSON + refusedJSON <- executeKirokuCommandWithStore store (previewCommand driftPath C.OutputJson) + assertOutcome (ExitFailure 2) "stream_head_drift" refusedJSON + assertJSON refusedJSON + refused <- executeKirokuCommandWithStore store (previewCommand driftPath C.OutputTable) + assertOutcome (ExitFailure 2) "CompactionStreamHeadDrift" refused + exit <- try (runKirokuCommandWithStore store (previewCommand driftPath C.OutputTable)) :: IO (Either ExitCode ()) + exit `shouldBe` Left (ExitFailure 2) + denied <- executeKirokuCommandWithStore store (applyCommand path (T.replicate 64 "0") C.OutputTable) + assertOutcome (ExitFailure 2) "does not match" denied + eventsBefore <- storeRead =<< runStoreIO store (readAllForward (GlobalPosition 0) 100) + V.length eventsBefore `shouldBe` 3 + first <- executeKirokuCommandWithStore store (applyCommand path digest C.OutputTable) + assertOutcome ExitSuccess "applied:" first + again <- executeKirokuCommandWithStore store (applyCommand path digest C.OutputTable) + assertOutcome ExitSuccess "already applied:" again + previewed <- executeKirokuCommandWithStore store (previewCommand path C.OutputTable) + assertOutcome ExitSuccess "already applied:" previewed + forM_ [applyCommand path digest C.OutputJson, previewCommand path C.OutputJson, C.KirokuCompaction (C.CompactionLedger (C.LedgerOptions 10 C.OutputJson))] $ \command -> do + result <- executeKirokuCommandWithStore store command + assertOutcome ExitSuccess "" result + assertJSON result + eventsAfter <- storeRead =<< runStoreIO store (readAllForward (GlobalPosition 0) 100) + map (\e -> e ^. #originalVersion) (V.toList eventsAfter) `shouldBe` [StreamVersion 1, StreamVersion 3] + records <- storeRead =<< runStoreIO store (compactionLedger (CompactionLedgerQuery (either (error . show) id (mkCompactionLedgerLimit 10)))) + V.length records `shouldBe` 1 + ledger <- executeKirokuCommandWithStore store (C.KirokuCompaction (C.CompactionLedger (C.LedgerOptions 10 C.OutputTable))) + assertOutcome ExitSuccess "MANIFEST_DIGEST" ledger + + it "rejects confirmation errors and invalid limits without any caller pool checkout" $ + withSystemTempDirectory "kiroku-confirm-manifest" $ \directory -> do + calls <- newIORef (0 :: Int) + owner <- myThreadId + withMigratedTestDatabase $ \conn -> do + let settings = + (defaultConnectionSettings conn) + { observationHandler = Just $ \case + ConnectionObservation _ InUseConnectionStatus -> do current <- myThreadId; when (current == owner) (modifyIORef' calls (+ 1)) + _ -> pure () + } + withStore settings $ \store -> do + let path = directory <> "/manifest.json" + Aeson.encodeFile path sampleManifest + initialCalls <- readIORef calls + mismatch <- executeKirokuCommandWithStore store (applyCommand path (T.replicate 64 "0") C.OutputJson) + assertOutcome (ExitFailure 2) "does not match" mismatch + assertJSON mismatch + unparseable <- executeKirokuCommandWithStore store (applyCommand path "invalid" C.OutputTable) + assertOutcome (ExitFailure 1) "invalid confirmation" unparseable + invalidLimit <- executeKirokuCommandWithStore store (C.KirokuCompaction (C.CompactionLedger (C.LedgerOptions 0 C.OutputJson))) + assertOutcome (ExitFailure 1) "ledger limit" invalidLimit + unreadable <- executeKirokuCommandWithStore store (previewCommand (directory <> "/missing") C.OutputJson) + assertOutcome (ExitFailure 1) "cannot read manifest" unreadable + assertJSON unreadable + readIORef calls `shouldReturn` initialCalls + +previewCommand :: FilePath -> C.OutputFormat -> C.KirokuCommand +previewCommand path format = C.KirokuCompaction (C.CompactionPreview (C.CompactionOptions (C.ManifestPath path) format)) + +applyCommand :: FilePath -> Text -> C.OutputFormat -> C.KirokuCommand +applyCommand path digest format = C.KirokuCompaction (C.CompactionApply (C.ApplyOptions (C.ManifestPath path) digest format)) + +assertOutcome :: ExitCode -> Text -> KirokuCommandOutcome -> Expectation +assertOutcome expected needle (KirokuCommandOutcome output actual) = do + actual `shouldBe` expected + output `shouldSatisfy` T.isInfixOf needle + +assertJSON :: KirokuCommandOutcome -> Expectation +assertJSON (KirokuCommandOutcome output _) = validJSON output + +validJSON :: Text -> Expectation +validJSON output = (Aeson.eitherDecodeStrict (TE.encodeUtf8 output) :: Either String Aeson.Value) `shouldSatisfy` isRight + +embeddableMessage :: Text +embeddableMessage = "kiroku: compaction and store identity commands run only inside a host that owns a KirokuStore; the standalone binary is a remote client" + +sampleManifest :: CompactionManifest +sampleManifest = either (error . show) id (mkCompactionManifest (CompactionManifestInput (StoreIdentity UUID.nil) op RefuseDeadLetters RefuseCausationDependents [StreamHeadWitness name (StreamVersion 1)] [CompactionSelection (EventId UUID.nil) name (StreamVersion 1) (GlobalPosition 1) V.empty] Nothing)) + where + name = StreamName "sample-1" + op = either (error . show) id (mkCompactionOperation "sample") + +withTestStore :: (KirokuStore -> IO a) -> IO a +withTestStore action = withMigratedTestDatabase $ \conn -> withStore (defaultConnectionSettings conn) action + +storeRead :: (Show e) => Either e a -> IO a +storeRead = either (fail . show) pure + +liveInput :: KirokuStore -> IO CompactionManifestInput +liveInput store = do + let name = StreamName "orders-cli" + event = EventData Nothing (EventType "CLI") (Aeson.object []) Nothing Nothing Nothing + _ <- storeRead =<< runStoreIO store (appendToStream name NoStream [event, event, event]) + events <- storeRead =<< runStoreIO store (readAllForward (GlobalPosition 0) 10) + identity <- storeRead =<< runStoreIO store storeIdentity + op <- either (fail . show) pure (mkCompactionOperation "cli-test") + let selected = events V.! 1 + pure (CompactionManifestInput identity op RefuseDeadLetters RefuseCausationDependents [StreamHeadWitness name (StreamVersion 3)] [CompactionSelection (selected ^. #eventId) name (selected ^. #originalVersion) (selected ^. #globalPosition) V.empty] Nothing)