llmgw/internal/kb/kbsync/sync_test.go

package kbsync

import (
	"context"
	"encoding/json"
	"errors"
	"net/http"
	"net/http/httptest"
	"os"
	"path/filepath"
	"strings"
	"sync"
	"testing"
	"time"
)

// fakeGateway holds one collection in memory and embeds by counting letters.
type fakeGateway struct {
	mu        sync.Mutex
	writable  bool
	docs      map[string]RemoteDoc
	puts      int
	embedded  int
	busyFirst bool
	// serverEmbeds is a gateway that embeds text it is sent; vectors
	// counts chunks that came with one.
	serverEmbeds bool
	vectors      int
}

func (g *fakeGateway) serve(t *testing.T) *Client {
	t.Helper()
	mux := http.NewServeMux()
	coll := Collection{ID: "kb_x", Name: "ik", EmbedModel: "emb", Dims: 3}
	mux.HandleFunc("GET /v1/kb/collections/{id}/manifest", func(w http.ResponseWriter, r *http.Request) {
		g.mu.Lock()
		defer g.mu.Unlock()
		if r.PathValue("id") != "ik" && r.PathValue("id") != "kb_x" {
			http.Error(w, `{"error":{"message":"no collection"}}`, http.StatusNotFound)
			return
		}
		m := Manifest{Collection: coll}
		m.Collection.Writable = g.writable
		m.Collection.ServerEmbeds = g.serverEmbeds
		for _, d := range g.docs {
			m.Documents = append(m.Documents, d)
		}
		_ = json.NewEncoder(w).Encode(m)
	})
	mux.HandleFunc("PUT /v1/kb/collections/{id}/documents", func(w http.ResponseWriter, r *http.Request) {
		var d DocOut
		_ = json.NewDecoder(r.Body).Decode(&d)
		g.mu.Lock()
		defer g.mu.Unlock()
		g.puts++
		for _, c := range d.Chunks {
			if len(c.Embedding) > 0 {
				g.vectors++
			}
		}
		g.docs[d.Name] = RemoteDoc{Name: d.Name, Digest: d.Digest, Chunks: len(d.Chunks)}
		_, _ = w.Write([]byte(`{"changed":true}`))
	})
	mux.HandleFunc("DELETE /v1/kb/collections/{id}/documents", func(w http.ResponseWriter, r *http.Request) {
		g.mu.Lock()
		defer g.mu.Unlock()
		delete(g.docs, r.URL.Query().Get("name"))
		_, _ = w.Write([]byte(`{}`))
	})
	mux.HandleFunc("POST /v1/embeddings", func(w http.ResponseWriter, r *http.Request) {
		g.mu.Lock()
		busy := g.busyFirst
		g.busyFirst = false
		g.mu.Unlock()
		if busy {
			w.Header().Set("Retry-After", "1")
			http.Error(w, `{"error":{"message":"the model is loading"}}`, http.StatusServiceUnavailable)
			return
		}
		var in struct {
			Input []string `json:"input"`
		}
		_ = json.NewDecoder(r.Body).Decode(&in)
		type row struct {
			Index     int       `json:"index"`
			Embedding []float32 `json:"embedding"`
		}
		var out struct {
			Data []row `json:"data"`
		}
		for i, s := range in.Input {
			out.Data = append(out.Data, row{i, []float32{float32(len(s)), 1, 0}})
		}
		g.mu.Lock()
		g.embedded += len(in.Input)
		g.mu.Unlock()
		_ = json.NewEncoder(w).Encode(out)
	})
	srv := httptest.NewServer(mux)
	t.Cleanup(srv.Close)
	h, _ := NewHTTP("")
	return &Client{Base: srv.URL, Key: "k", HTTP: h}
}

func setup(t *testing.T) (*fakeGateway, *Client, *Ledger, string) {
	t.Helper()
	g := &fakeGateway{writable: true, docs: map[string]RemoteDoc{}}
	c := g.serve(t)
	l, err := OpenLedger(context.Background(), filepath.Join(t.TempDir(), "ledger.db"))
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = l.Close() })
	dir := t.TempDir()
	write(t, dir, "politika/izin.md", "# İzin\n\nYirmi gün.\n")
	write(t, dir, "politika/masraf.txt", "Masraflar finansa.\n")
	write(t, dir, "eski.doc", "\xd0\xcf\x11\xe0\x00binary")
	write(t, dir, "resim.bin", "\x00\x01")
	write(t, dir, ".git/config", "[core]")
	write(t, dir, "politika/~$izin.docx", "lock")
	return g, c, l, dir
}

func write(t *testing.T, dir, name, s string) {
	t.Helper()
	p := filepath.Join(dir, filepath.FromSlash(name))
	if err := os.MkdirAll(filepath.Dir(p), 0o755); err != nil {
		t.Fatal(err)
	}
	if err := os.WriteFile(p, []byte(s), 0o644); err != nil {
		t.Fatal(err)
	}
}

func kinds(events []Event) map[string][]string {
	out := map[string][]string{}
	for _, e := range events {
		out[e.Kind] = append(out[e.Kind], e.Name)
	}
	return out
}

func run(t *testing.T, c *Client, l *Ledger, o Options) (Report, map[string][]string) {
	t.Helper()
	var mu sync.Mutex
	var ev []Event
	rep, err := Sync(context.Background(), c, l, o, func(e Event) {
		mu.Lock()
		ev = append(ev, e)
		mu.Unlock()
	})
	if err != nil {
		t.Fatal(err)
	}
	return rep, kinds(ev)
}

func TestASecondRunSendsNothingAndReadsNothing(t *testing.T) {
	g, c, l, dir := setup(t)
	rep, k := run(t, c, l, Options{Collection: "ik", Roots: []string{dir}})
	if rep.Sent != 2 || len(g.docs) != 2 {
		t.Fatalf("first run %+v %v", rep, k)
	}
	if _, ok := g.docs["politika/izin.md"]; !ok {
		t.Errorf("documents %v", g.docs)
	}
	// The legacy Word file is skipped with the reason, and no hidden,
	// lock or unknown file is touched.
	if len(k["skipped"]) != 1 || k["skipped"][0] != "eski.doc" {
		t.Errorf("skipped %v", k["skipped"])
	}

	puts, embedded := g.puts, g.embedded
	rep, k = run(t, c, l, Options{Collection: "ik", Roots: []string{dir}})
	if rep.Sent != 0 || rep.Unchanged != 2 || g.puts != puts || g.embedded != embedded {
		t.Errorf("second run %+v %v", rep, k)
	}
	if rep.Skipped != 1 {
		t.Errorf("the skipped file was not reported as before: %+v", rep)
	}
	es, _ := l.Entries(context.Background(), c.Base, "kb_x")
	if len(es) != 3 {
		t.Errorf("ledger %+v", es)
	}
	runs, _ := l.Runs(context.Background(), 10)
	if len(runs) != 2 || runs[0].Unchanged != 2 {
		t.Errorf("runs %+v", runs)
	}
}

func TestAChangedFileIsSentAgain(t *testing.T) {
	g, c, l, dir := setup(t)
	run(t, c, l, Options{Collection: "ik", Roots: []string{dir}})
	old := g.docs["politika/izin.md"].Digest
	write(t, dir, "politika/izin.md", "# İzin\n\nOtuz gün.\n")
	// A changed time is what makes it look again; some filesystems keep
	// seconds only.
	later := time.Now().Add(2 * time.Second)
	_ = os.Chtimes(filepath.Join(dir, "politika", "izin.md"), later, later)
	rep, k := run(t, c, l, Options{Collection: "ik", Roots: []string{dir}})
	if rep.Sent != 1 || k["sent"][0] != "politika/izin.md" || g.docs["politika/izin.md"].Digest == old {
		t.Errorf("after a change %+v %v", rep, k)
	}
}

func TestDeleteRemovesOnlyWhatThisMachineSentUnderTheseRoots(t *testing.T) {
	g, c, l, dir := setup(t)
	run(t, c, l, Options{Collection: "ik", Roots: []string{dir}})
	// Somebody else's document, sent from another machine.
	g.docs["baska/makine.md"] = RemoteDoc{Name: "baska/makine.md", Digest: "x"}
	// A file of this machine from another folder.
	other := t.TempDir()
	write(t, other, "not.md", "başka klasör")
	run(t, c, l, Options{Collection: "ik", Roots: []string{other}})

	if err := os.Remove(filepath.Join(dir, "politika", "masraf.txt")); err != nil {
		t.Fatal(err)
	}
	rep, k := run(t, c, l, Options{Collection: "ik", Roots: []string{dir}, Delete: true})
	if rep.Deleted != 1 || k["deleted"][0] != "politika/masraf.txt" {
		t.Errorf("deleted %+v %v", rep, k)
	}
	for _, keep := range []string{"baska/makine.md", "not.md", "politika/izin.md"} {
		if _, ok := g.docs[keep]; !ok {
			t.Errorf("%s was deleted", keep)
		}
	}
}

func TestADryRunSendsAndRecordsNothing(t *testing.T) {
	g, c, l, dir := setup(t)
	rep, k := run(t, c, l, Options{Collection: "ik", Roots: []string{dir}, DryRun: true})
	if rep.Sent != 2 || len(k["would-send"]) != 2 || rep.Bytes == 0 || rep.Chunks != 2 {
		t.Errorf("dry run %+v %v", rep, k)
	}
	if g.puts != 0 || g.embedded != 0 {
		t.Errorf("a dry run sent %d documents, embedded %d chunks", g.puts, g.embedded)
	}
	if es, _ := l.Entries(context.Background(), c.Base, "kb_x"); len(es) != 0 {
		t.Errorf("a dry run wrote the ledger: %+v", es)
	}
}

func TestAReadOnlyCollectionIsRefusedAndALoadingModelWaitedFor(t *testing.T) {
	g, c, l, dir := setup(t)
	g.writable = false
	if _, err := Sync(context.Background(), c, l, Options{Collection: "ik", Roots: []string{dir}}, nil); err == nil ||
		!strings.Contains(err.Error(), "not add") {
		t.Errorf("read-only gave %v", err)
	}
	g.writable, g.busyFirst = true, true
	var waited []string
	c.Waiting = func(why string, _ time.Duration) { waited = append(waited, why) }
	rep, _ := run(t, c, l, Options{Collection: "ik", Roots: []string{dir}, Workers: 1})
	if rep.Sent != 2 || len(waited) != 1 || !strings.Contains(waited[0], "loading") {
		t.Errorf("after a 503 %+v, waited %v", rep, waited)
	}
	if _, err := Sync(context.Background(), c, l, Options{Collection: "yok", Roots: []string{dir}}, nil); err == nil {
		t.Error("a collection that is not there was synced")
	}
}

func TestFiltersAndNames(t *testing.T) {
	_, c, l, dir := setup(t)
	write(t, dir, "rapor/2026/ocak.md", "ocak")
	write(t, dir, "rapor/taslak.md", "taslak")
	rep, k := run(t, c, l, Options{Collection: "ik", Roots: []string{dir}, DryRun: true,
		Include: []string{"rapor/**/*.md"}, Exclude: []string{"taslak*"}})
	if rep.Sent != 1 || k["would-send"][0] != "rapor/2026/ocak.md" {
		t.Errorf("filtered %+v %v", rep, k)
	}
	// Two roots keep their names apart.
	other := t.TempDir()
	write(t, other, "politika/izin.md", "aynı ad")
	_, k = run(t, c, l, Options{Collection: "ik", Roots: []string{dir, other}, DryRun: true, Include: []string{"izin.md"}})
	if len(k["would-send"]) != 2 || k["would-send"][0] == k["would-send"][1] {
		t.Errorf("two roots: %v", k["would-send"])
	}
	if _, err := Sync(context.Background(), c, l, Options{Collection: "ik", Roots: []string{t.TempDir()}}, nil); !errors.Is(err, ErrNothing) {
		t.Errorf("an empty folder gave %v", err)
	}
}

func TestAGatewayThatEmbedsIsSentTextAlone(t *testing.T) {
	g, c, l, dir := setup(t)
	g.serverEmbeds = true
	rep, k := run(t, c, l, Options{Collection: "ik", Roots: []string{dir}})
	if len(k["sent"]) != 2 || rep.Failed != 0 {
		t.Fatalf("%+v %v", rep, k)
	}
	if g.embedded != 0 || g.vectors != 0 {
		t.Errorf("embedded here: %d asked, %d vectors sent", g.embedded, g.vectors)
	}

	// --embed-here sends the vectors all the same.
	write(t, dir, "politika/yeni.md", "Yeni kural.\n")
	run(t, c, l, Options{Collection: "ik", Roots: []string{dir}, EmbedHere: true})
	if g.embedded == 0 || g.vectors == 0 {
		t.Errorf("--embed-here: %d asked, %d vectors sent", g.embedded, g.vectors)
	}
}