Update 1.5.0
release-tag / release-image (push) Successful in 10m52s

This commit is contained in:
2026-08-27 07:51:39 +02:00
parent 1decb831d6
commit 8c67c7a7fa
58 changed files with 10768 additions and 626 deletions
+4 -1
View File
@@ -3,7 +3,9 @@ WORKDIR /src
COPY go.mod ./
COPY cmd ./cmd
COPY internal ./internal
RUN CGO_ENABLED=0 GOOS=linux go build -trimpath -ldflags="-s -w" -o /out/glpi-ai-agent ./cmd/agent
RUN go test ./... && \
go vet ./... && \
CGO_ENABLED=0 GOOS=linux go build -trimpath -ldflags="-s -w" -o /out/glpi-ai-agent ./cmd/agent
# One-shot helper used by docker compose to prepare the persistent volume for
# the distroless non-root runtime user (UID/GID 65532).
@@ -16,5 +18,6 @@ COPY --from=build /out/glpi-ai-agent /app/glpi-ai-agent
COPY knowledge /app/knowledge
VOLUME ["/app/data"]
EXPOSE 8080
HEALTHCHECK --interval=15s --timeout=3s --start-period=10s --retries=4 CMD ["/app/glpi-ai-agent", "healthcheck"]
USER 65532:65532
ENTRYPOINT ["/app/glpi-ai-agent"]
+20
View File
@@ -27,6 +27,10 @@ import (
)
func main() {
if len(os.Args) == 2 && os.Args[1] == "healthcheck" {
runHealthcheck()
return
}
cfg, err := config.Load()
if err != nil {
slog.Error("configuration invalid", "error", err)
@@ -204,6 +208,22 @@ func main() {
slog.Info("shutdown complete")
}
func runHealthcheck() {
client := &http.Client{Timeout: 2 * time.Second}
req, err := http.NewRequest(http.MethodGet, "http://127.0.0.1:8080/healthz", nil)
if err != nil {
os.Exit(1)
}
resp, err := client.Do(req)
if err != nil {
os.Exit(1)
}
defer resp.Body.Close()
if resp.StatusCode < 200 || resp.StatusCode >= 300 {
os.Exit(1)
}
}
func waitForOllamaPool(ctx context.Context, client *ollama.Client, retryInterval time.Duration) error {
if retryInterval < 2*time.Second {
retryInterval = 5 * time.Second
+7
View File
@@ -325,6 +325,7 @@ func Load() (Config, error) {
CommunicationSalutation: env("COMMUNICATION_SALUTATION", "Guten Tag,"),
CommunicationClosing: env("COMMUNICATION_CLOSING", "Mit freundlichen Grüßen"),
CommunicationSignature: env("COMMUNICATION_SIGNATURE", "IT-Service"),
AIContentLabelEnabled: envBool("AI_CONTENT_LABEL_ENABLED", true),
AutoCategory: envBool("AUTO_CATEGORY", true),
AutoReply: envBool("AUTO_REPLY", false),
PriorityEnabled: envBool("PRIORITY_ENABLED", true),
@@ -651,6 +652,12 @@ func (c Config) Validate() error {
return fmt.Errorf("KNOWLEDGE_VECTOR_BACKEND must be one of: local, dual, neuroforge (got %q)", c.KnowledgeVectorBackend)
}
if c.KnowledgeVectorBackend == "dual" || c.KnowledgeVectorBackend == "neuroforge" {
if len(c.NeuroForgeAPIKey) < 24 {
return errors.New("NEUROFORGE_API_KEY must contain at least 24 characters for dual/neuroforge mode")
}
if isPlaceholder(c.NeuroForgeAPIKey) {
return errors.New("NEUROFORGE_API_KEY still contains a CHANGE_ME placeholder")
}
u, err := url.Parse(c.NeuroForgeURL)
if err != nil || u.Scheme == "" || u.Host == "" || (u.Scheme != "http" && u.Scheme != "https") {
return errors.New("NEUROFORGE_URL must be an absolute http(s) URL")
@@ -515,3 +515,40 @@ func TestValidateGLPIKBAutoReplyUncategorizedRequiresArticleAllowlist(t *testing
t.Fatalf("expected uncategorized auto reply with explicit article allowlist to validate: %v", err)
}
}
func TestLoadReadsAIContentLabelEnabled(t *testing.T) {
for key, value := range map[string]string{
"GLPI_URL": "https://glpi.internal.example",
"GLPI_CLIENT_ID": "client-id",
"GLPI_CLIENT_SECRET": "real-secret-value",
"GLPI_USERNAME": "svc-agent",
"GLPI_PASSWORD": "real-password-value",
"WEB_ALLOW_ANONYMOUS": "true",
"AI_CONTENT_LABEL_ENABLED": "false",
} {
t.Setenv(key, value)
}
c, err := Load()
if err != nil {
t.Fatal(err)
}
if c.AIContentLabelEnabled {
t.Fatal("AI_CONTENT_LABEL_ENABLED=false was not loaded")
}
}
func TestValidateNeuroForgeAPIKeyForRemoteVectorBackend(t *testing.T) {
c := validConfig()
c.KnowledgeVectorBackend = "dual"
c.NeuroForgeURL = "http://neuroforge:8090"
c.NeuroForgeSearchK = 128
c.NeuroForgeTimeout = 5 * time.Second
c.NeuroForgeAPIKey = "CHANGE_ME_NEUROFORGE_API_KEY"
if err := c.Validate(); err == nil {
t.Fatal("expected placeholder NEUROFORGE_API_KEY to be rejected")
}
c.NeuroForgeAPIKey = "integration-token-0123456789abcdef"
if err := c.Validate(); err != nil {
t.Fatalf("expected strong NEUROFORGE_API_KEY to validate: %v", err)
}
}
+3 -1
View File
@@ -1,7 +1,9 @@
FROM golang:1.23-alpine AS build
WORKDIR /src
COPY . .
RUN CGO_ENABLED=0 go build -trimpath -ldflags="-s -w" -o /out/control .
RUN go test ./... && \
go vet ./... && \
CGO_ENABLED=0 go build -trimpath -ldflags="-s -w" -o /out/control .
FROM alpine:3.21
RUN addgroup -S app && adduser -S -G app app
COPY --from=build /out/control /usr/local/bin/control
File diff suppressed because it is too large Load Diff
+86 -2
View File
@@ -2,14 +2,18 @@ package main
import (
"context"
"crypto/subtle"
"embed"
"encoding/json"
"errors"
"fmt"
"io"
"log"
"net/http"
"os"
"os/signal"
"strings"
"syscall"
"time"
)
@@ -70,6 +74,11 @@ func env(k, d string) string {
}
func main() {
if err := validateControlSecrets(); err != nil {
log.Fatal(err)
}
ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
defer stop()
agentURL := env("AGENT_URL", "http://agent:8080")
nfURL := env("NEUROFORGE_URL", "http://neuroforge:8080")
nfKeyRaw := strings.TrimSpace(os.Getenv("NEUROFORGE_API_KEY"))
@@ -109,9 +118,84 @@ func main() {
_, _ = w.Write(b)
})
addr := env("CONTROL_ADDR", ":8070")
srv := &http.Server{Addr: addr, Handler: secure(mux), ReadHeaderTimeout: 5 * time.Second, ReadTimeout: 10 * time.Second, WriteTimeout: 10 * time.Second, IdleTimeout: 60 * time.Second}
srv := &http.Server{Addr: addr, Handler: secure(controlBasicAuth(mux)), ReadHeaderTimeout: 5 * time.Second, ReadTimeout: 10 * time.Second, WriteTimeout: 10 * time.Second, IdleTimeout: 60 * time.Second}
log.Printf("mega control listening on %s", addr)
log.Fatal(srv.ListenAndServe())
errCh := make(chan error, 1)
go func() {
err := srv.ListenAndServe()
if errors.Is(err, http.ErrServerClosed) {
err = nil
}
errCh <- err
}()
select {
case err := <-errCh:
if err != nil {
log.Printf("control HTTP server stopped unexpectedly: %v", err)
}
case <-ctx.Done():
log.Printf("control shutdown requested")
}
shutdownCtx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
defer cancel()
if err := srv.Shutdown(shutdownCtx); err != nil {
log.Printf("control graceful shutdown failed: %v", err)
_ = srv.Close()
}
}
func validateControlSecrets() error {
check := func(name string, min int) error {
v := strings.TrimSpace(os.Getenv(name))
if v == "" {
return nil
}
u := strings.ToUpper(v)
if strings.Contains(u, "CHANGE_ME") || strings.Contains(u, "CHANGEME") || strings.Contains(u, "PLACEHOLDER") {
return fmt.Errorf("%s still contains a placeholder", name)
}
if len(v) < min {
return fmt.Errorf("%s must be at least %d characters", name, min)
}
return nil
}
if err := check("NEUROFORGE_API_KEY", 24); err != nil {
return err
}
if err := check("CONTROL_READ_TOKEN", 24); err != nil {
return err
}
if err := check("CONTROL_BASIC_AUTH_PASSWORD", 12); err != nil {
return err
}
u, p := strings.TrimSpace(os.Getenv("CONTROL_BASIC_AUTH_USER")), strings.TrimSpace(os.Getenv("CONTROL_BASIC_AUTH_PASSWORD"))
if (u == "") != (p == "") {
return errors.New("CONTROL_BASIC_AUTH_USER and CONTROL_BASIC_AUTH_PASSWORD must both be set or both be empty")
}
return nil
}
func controlBasicAuth(next http.Handler) http.Handler {
user := strings.TrimSpace(os.Getenv("CONTROL_BASIC_AUTH_USER"))
pass := strings.TrimSpace(os.Getenv("CONTROL_BASIC_AUTH_PASSWORD"))
if user == "" && pass == "" {
return next
}
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.URL.Path == "/healthz" {
next.ServeHTTP(w, r)
return
}
u, p, ok := r.BasicAuth()
userOK := subtle.ConstantTimeCompare([]byte(u), []byte(user)) == 1
passOK := subtle.ConstantTimeCompare([]byte(p), []byte(pass)) == 1
if !ok || !userOK || !passOK {
w.Header().Set("WWW-Authenticate", `Basic realm="NeuroForge Control", charset="UTF-8"`)
http.Error(w, "authentication required", http.StatusUnauthorized)
return
}
next.ServeHTTP(w, r)
})
}
func secure(next http.Handler) http.Handler {
+43
View File
@@ -0,0 +1,43 @@
package main
import (
"net/http"
"net/http/httptest"
"testing"
)
func TestControlBasicAuthProtectsOperationsButNotHealth(t *testing.T) {
t.Setenv("CONTROL_BASIC_AUTH_USER", "operator")
t.Setenv("CONTROL_BASIC_AUTH_PASSWORD", "control-password-123456")
h := controlBasicAuth(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusOK) }))
health := httptest.NewRecorder()
h.ServeHTTP(health, httptest.NewRequest(http.MethodGet, "/healthz", nil))
if health.Code != http.StatusOK {
t.Fatalf("health status=%d", health.Code)
}
unauth := httptest.NewRecorder()
h.ServeHTTP(unauth, httptest.NewRequest(http.MethodGet, "/api/status", nil))
if unauth.Code != http.StatusUnauthorized {
t.Fatalf("unauth status=%d", unauth.Code)
}
req := httptest.NewRequest(http.MethodGet, "/api/status", nil)
req.SetBasicAuth("operator", "control-password-123456")
auth := httptest.NewRecorder()
h.ServeHTTP(auth, req)
if auth.Code != http.StatusOK {
t.Fatalf("auth status=%d", auth.Code)
}
}
func TestControlSecretValidationRejectsPlaceholder(t *testing.T) {
t.Setenv("NEUROFORGE_API_KEY", "valid-neuroforge-read-token-123456")
t.Setenv("CONTROL_READ_TOKEN", "valid-agent-control-token-123456")
t.Setenv("CONTROL_BASIC_AUTH_USER", "operator")
t.Setenv("CONTROL_BASIC_AUTH_PASSWORD", "CHANGE_ME_CONTROL_PASSWORD")
if err := validateControlSecrets(); err == nil {
t.Fatal("placeholder password accepted")
}
}
+3 -1
View File
@@ -3,7 +3,9 @@ WORKDIR /src
COPY go.mod ./
COPY cmd ./cmd
COPY internal ./internal
RUN CGO_ENABLED=0 GOOS=linux go build -trimpath -ldflags="-s -w" -o /out/kb-helpdesk ./cmd/server
RUN go test ./... && \
go vet ./... && \
CGO_ENABLED=0 GOOS=linux go build -trimpath -ldflags="-s -w" -o /out/kb-helpdesk ./cmd/server
FROM alpine:3.24
RUN apk add --no-cache ca-certificates tzdata
+44 -2
View File
@@ -9,6 +9,7 @@ import (
"io"
"io/fs"
"net/http"
"net/url"
"os"
"strconv"
"strings"
@@ -72,6 +73,7 @@ func (a *app) routes() http.Handler {
mux.HandleFunc("GET /api/staging", a.handleStagingList)
mux.HandleFunc("GET /api/staging/{key}", a.handleStagingGet)
mux.HandleFunc("POST /api/integrations/staging", a.handleIntegrationStaging)
mux.HandleFunc("GET /api/integrations/staging/health", a.handleIntegrationStagingHealth)
if a.config.Writable {
mux.HandleFunc("PUT /api/items/{key}", a.handlePut)
@@ -93,7 +95,30 @@ func (a *app) routes() http.Handler {
static := http.FileServer(http.FS(a.web))
mux.Handle("GET /", static)
return securityHeaders(mux)
return securityHeaders(browserWriteSameOrigin(mux))
}
func browserWriteSameOrigin(next http.Handler) http.Handler {
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.Method == http.MethodGet || r.Method == http.MethodHead || r.Method == http.MethodOptions || r.URL.Path == "/api/integrations/staging" {
next.ServeHTTP(w, r)
return
}
fetchSite := strings.ToLower(strings.TrimSpace(r.Header.Get("Sec-Fetch-Site")))
if fetchSite != "" && fetchSite != "same-origin" && fetchSite != "none" {
writeError(w, http.StatusForbidden, "cross-origin browser write blocked")
return
}
origin := strings.TrimSpace(r.Header.Get("Origin"))
if origin != "" {
u, err := url.Parse(origin)
if err != nil || !strings.EqualFold(u.Host, r.Host) {
writeError(w, http.StatusForbidden, "cross-origin browser write blocked")
return
}
}
next.ServeHTTP(w, r)
})
}
func securityHeaders(next http.Handler) http.Handler {
@@ -265,6 +290,19 @@ func integrationBearerAuthorized(r *http.Request) (bool, bool) {
return true, subtle.ConstantTimeCompare([]byte(provided), []byte(expected)) == 1
}
func (a *app) handleIntegrationStagingHealth(w http.ResponseWriter, r *http.Request) {
enabled, authorized := integrationBearerAuthorized(r)
if !enabled || a.staging == nil {
writeError(w, http.StatusServiceUnavailable, "KB staging integration is disabled")
return
}
if !authorized {
writeError(w, http.StatusUnauthorized, "invalid integration token")
return
}
writeJSON(w, http.StatusOK, map[string]any{"ok": true, "staging": true, "writable": a.config.Writable})
}
// handleIntegrationStaging is a one-way governance boundary: machine-generated
// research may enter human review, but it cannot write production knowledge or
// enable automatic replies.
@@ -485,7 +523,11 @@ func (a *app) promoteStaging(key string) (map[string]any, error) {
}
archive, err := a.staging.ArchiveApproved(key)
if err != nil {
return nil, fmt.Errorf("Produktivdatei wurde erstellt (%s), aber Staging konnte nicht als freigegeben archiviert werden: %w", summary.RelPath, err)
rollbackErr := a.store.RollbackImported(summary)
if rollbackErr != nil {
return nil, fmt.Errorf("staging archive failed after production import (%s); rollback also failed: archive=%v rollback=%v", summary.RelPath, err, rollbackErr)
}
return nil, fmt.Errorf("staging archive failed; production import %s was rolled back: %w", summary.RelPath, err)
}
return map[string]any{"ok": true, "production": summary, "staging_key": key, "staging_archive": archive}, nil
}
+90
View File
@@ -459,3 +459,93 @@ func TestIntegrationDraftWithStableKeyUpdatesInsteadOfDuplicating(t *testing.T)
t.Fatalf("draft not refreshed: %#v", got.Document)
}
}
func TestBrowserWriteSameOriginGuardRejectsCrossSiteWrite(t *testing.T) {
s, err := store.New(t.TempDir())
if err != nil {
t.Fatal(err)
}
web, err := fs.Sub(webFS, "web")
if err != nil {
t.Fatal(err)
}
h := newApp(s, web, appConfig{Mode: "editor", Writable: true}).routes()
req := httptest.NewRequest(http.MethodPost, "/api/bulk", bytes.NewBufferString(`{"keys":[],"dry_run":true}`))
req.Header.Set("Content-Type", "application/json")
req.Header.Set("Sec-Fetch-Site", "cross-site")
req.Header.Set("Origin", "https://evil.invalid")
rr := httptest.NewRecorder()
h.ServeHTTP(rr, req)
if rr.Code != http.StatusForbidden {
t.Fatalf("status=%d body=%s", rr.Code, rr.Body.String())
}
}
func TestPromotionRollsBackProductionWhenStagingArchiveFails(t *testing.T) {
knowledgeDir := t.TempDir()
stagingDir := t.TempDir()
t.Setenv("BACKUP_DIR", filepath.Join(t.TempDir(), "backups"))
s, err := store.New(knowledgeDir)
if err != nil {
t.Fatal(err)
}
st, err := staging.New(stagingDir)
if err != nil {
t.Fatal(err)
}
draft, err := st.Save("rollback", "test", staging.Draft{Title: "Rollback", Text: "Symptom", Answer: "Lösung"}, false, .8)
if err != nil {
t.Fatal(err)
}
// Force ArchiveApproved to fail after ImportDocument by occupying the archive
// directory path with a regular file.
if err := os.WriteFile(filepath.Join(stagingDir, ".approved"), []byte("block"), 0o644); err != nil {
t.Fatal(err)
}
web, err := fs.Sub(webFS, "web")
if err != nil {
t.Fatal(err)
}
a := newApp(s, web, appConfig{Mode: "editor", Writable: true}).withStaging(st)
if _, err := a.promoteStaging(draft.Key); err == nil || !strings.Contains(err.Error(), "rolled back") {
t.Fatalf("promotion error=%v", err)
}
if s.Count() != 0 {
t.Fatalf("production count=%d, want rollback to zero", s.Count())
}
if _, err := st.Get(draft.Key); err != nil {
t.Fatalf("staging draft should remain for retry: %v", err)
}
}
func TestIntegrationStagingHealthBypassesUIBasicAuthButRequiresBearer(t *testing.T) {
t.Setenv("BASIC_AUTH_USER", "editor")
t.Setenv("BASIC_AUTH_PASSWORD", "knowledge-password-123456")
t.Setenv("KB_INTEGRATION_TOKEN", "integration-token-12345678901234567890")
s, err := store.New(t.TempDir())
if err != nil {
t.Fatal(err)
}
st, err := staging.New(t.TempDir())
if err != nil {
t.Fatal(err)
}
web, err := fs.Sub(webFS, "web")
if err != nil {
t.Fatal(err)
}
h := optionalBasicAuth(newApp(s, web, appConfig{Mode: "editor", Writable: true}).withStaging(st).routes())
unauth := httptest.NewRecorder()
h.ServeHTTP(unauth, httptest.NewRequest(http.MethodGet, "/api/integrations/staging/health", nil))
if unauth.Code != http.StatusUnauthorized || strings.Contains(unauth.Body.String(), "authentication required") {
t.Fatalf("request should reach bearer guard, status=%d body=%s", unauth.Code, unauth.Body.String())
}
req := httptest.NewRequest(http.MethodGet, "/api/integrations/staging/health", nil)
req.Header.Set("Authorization", "Bearer integration-token-12345678901234567890")
rr := httptest.NewRecorder()
h.ServeHTTP(rr, req)
if rr.Code != http.StatusOK {
t.Fatalf("status=%d body=%s", rr.Code, rr.Body.String())
}
}
+72 -8
View File
@@ -1,17 +1,21 @@
package main
import (
"context"
"crypto/subtle"
"embed"
"errors"
"flag"
"fmt"
"io/fs"
"log"
"net/http"
"os"
"os/signal"
"path/filepath"
"strconv"
"strings"
"syscall"
"time"
"kb-editor/internal/aifallback"
@@ -29,6 +33,12 @@ func main() {
flag.StringVar(&listen, "listen", envOr("LISTEN_ADDR", ":8080"), "HTTP listen address")
flag.Parse()
ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
defer stop()
if err := validateRuntimeSecrets(); err != nil {
log.Fatal(err)
}
cfg, staticDir, err := configFromEnv()
if err != nil {
log.Fatal(err)
@@ -58,7 +68,7 @@ func main() {
log.Fatal(err)
}
if reloadInterval > 0 {
go startAutoReload(s, reloadInterval)
go startAutoReload(ctx, s, reloadInterval)
}
sub, err := fs.Sub(webFS, staticDir)
@@ -95,9 +105,58 @@ func main() {
if u := os.Getenv("BASIC_AUTH_USER"); u != "" {
log.Printf("Basic authentication enabled for user %q", u)
}
if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed {
log.Fatal(err)
errCh := make(chan error, 1)
go func() {
err := srv.ListenAndServe()
if errors.Is(err, http.ErrServerClosed) {
err = nil
}
errCh <- err
}()
select {
case err := <-errCh:
if err != nil {
log.Printf("KB HTTP server stopped unexpectedly: %v", err)
}
case <-ctx.Done():
log.Printf("KB shutdown requested")
}
shutdownCtx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()
if err := srv.Shutdown(shutdownCtx); err != nil {
log.Printf("KB graceful shutdown failed: %v", err)
_ = srv.Close()
}
}
func validateRuntimeSecrets() error {
check := func(name string, min int) error {
v := strings.TrimSpace(os.Getenv(name))
if v == "" {
return nil
}
upper := strings.ToUpper(v)
if strings.Contains(upper, "CHANGE_ME") || strings.Contains(upper, "CHANGEME") || strings.Contains(upper, "PLACEHOLDER") {
return fmt.Errorf("%s still contains a placeholder", name)
}
if len(v) < min {
return fmt.Errorf("%s must be at least %d characters", name, min)
}
return nil
}
for _, item := range []struct {
name string
min int
}{
{"KB_INTEGRATION_TOKEN", 24},
{"BASIC_AUTH_PASSWORD", 12},
{"BRAIN_ACTIVITY_API_KEY", 24},
} {
if err := check(item.name, item.min); err != nil {
return err
}
}
return nil
}
func configFromEnv() (appConfig, string, error) {
@@ -143,12 +202,17 @@ func autoReloadInterval(mode string) (time.Duration, error) {
return d, nil
}
func startAutoReload(s *store.Store, interval time.Duration) {
func startAutoReload(ctx context.Context, s *store.Store, interval time.Duration) {
ticker := time.NewTicker(interval)
defer ticker.Stop()
for range ticker.C {
if err := s.Reload(); err != nil {
log.Printf("automatic index reload failed: %v", err)
for {
select {
case <-ctx.Done():
return
case <-ticker.C:
if err := s.Reload(); err != nil {
log.Printf("automatic index reload failed: %v", err)
}
}
}
}
@@ -254,7 +318,7 @@ func optionalBasicAuth(next http.Handler) http.Handler {
log.Fatal("BASIC_AUTH_USER and BASIC_AUTH_PASSWORD must either both be set or both be empty")
}
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if (r.Method == http.MethodGet && r.URL.Path == "/api/health") || (r.Method == http.MethodPost && r.URL.Path == "/api/integrations/staging") {
if (r.Method == http.MethodGet && (r.URL.Path == "/api/health" || r.URL.Path == "/api/integrations/staging/health")) || (r.Method == http.MethodPost && r.URL.Path == "/api/integrations/staging") {
next.ServeHTTP(w, r)
return
}
+28 -1
View File
@@ -377,6 +377,24 @@ func (s *Store) archive(key, bucket string) (string, error) {
if err := os.Rename(path, dst); err != nil {
return "", fmt.Errorf("move staging file to %s: %w", bucket, err)
}
rollbackRename := func(cause error) error {
if err := os.Rename(dst, path); err != nil {
return fmt.Errorf("%v; staging archive rollback failed: %w", cause, err)
}
if err := syncDir(s.dir); err != nil {
return fmt.Errorf("%v; staging rollback source sync failed: %w", cause, err)
}
if err := syncDir(archiveDir); err != nil {
return fmt.Errorf("%v; staging rollback archive sync failed: %w", cause, err)
}
return cause
}
if err := syncDir(s.dir); err != nil {
return "", rollbackRename(fmt.Errorf("sync staging directory: %w", err))
}
if err := syncDir(archiveDir); err != nil {
return "", rollbackRename(fmt.Errorf("sync staging archive: %w", err))
}
return dst, nil
}
@@ -433,7 +451,16 @@ func atomicWrite(path string, payload []byte, mode os.FileMode) error {
if err := os.Rename(tmpName, path); err != nil {
return err
}
return nil
return syncDir(filepath.Dir(path))
}
func syncDir(dir string) error {
f, err := os.Open(dir)
if err != nil {
return err
}
defer f.Close()
return f.Sync()
}
func summarize(result Result) Summary {
@@ -686,6 +686,10 @@ func (s *Store) ImportDocument(doc map[string]any, preferredBase string) (Summar
_ = os.Remove(tmpName)
return Summary{}, err
}
if err := syncDir(s.dataDir); err != nil {
_ = os.Remove(path)
return Summary{}, err
}
rec, err := s.readRecord(path)
if err != nil {
_ = os.Remove(path)
@@ -697,6 +701,47 @@ func (s *Store) ImportDocument(doc map[string]any, preferredBase string) (Summar
return summarize(rec), nil
}
// RollbackImported removes exactly the production record created by ImportDocument.
// The checksum guard prevents rollback from deleting a file that was changed after
// import. It is used to keep staging promotion transactional.
func (s *Store) RollbackImported(summary Summary) error {
s.mu.Lock()
defer s.mu.Unlock()
rec, ok := s.records[summary.Key]
if !ok {
return os.ErrNotExist
}
if summary.Checksum == "" || rec.Checksum != summary.Checksum {
return fmt.Errorf("refusing rollback: production record %s changed after import", summary.RelPath)
}
if err := os.Remove(rec.Path); err != nil {
return err
}
// The namespace mutation already happened once Remove succeeded. Reflect it
// in memory even when the durability fsync below reports an I/O error;
// otherwise this process would serve a record whose file no longer exists.
delete(s.records, summary.Key)
for i, key := range s.order {
if key == summary.Key {
s.order = append(s.order[:i], s.order[i+1:]...)
break
}
}
if err := syncDir(s.dataDir); err != nil {
return fmt.Errorf("sync production directory after rollback: %w", err)
}
return nil
}
func syncDir(dir string) error {
f, err := os.Open(dir)
if err != nil {
return err
}
defer f.Close()
return f.Sync()
}
func safeFilenameBase(value string) string {
value = strings.TrimSpace(value)
if value == "" {
@@ -947,6 +992,9 @@ func (s *Store) writeRecord(rec *record, doc map[string]any) (*record, error) {
_ = os.Remove(tmpName)
return nil, err
}
if err := syncDir(filepath.Dir(rec.Path)); err != nil {
return nil, err
}
return s.readRecord(rec.Path)
}