tatara-memory-repo-ingester Implementation Plan¶

// Package version holds build-time version information populated via ldflags.
package version

// Version is the semantic version string, set at build time.
var Version = "dev"

// Commit is the git commit SHA, set at build time.
var Commit = "unknown"

// Date is the build timestamp in RFC3339 format, set at build time.
var Date = "unknown"

// Command tatara-ingest walks a repository and pushes its code graph and
// semantic chunks to tatara-memory.
package main

import "fmt"

func main() {
    fmt.Println("tatara-ingest")
}

package contract_test

import (
    "encoding/json"
    "testing"

    "github.com/stretchr/testify/require"

    "github.com/szymonrychu/tatara-memory-repo-ingester/internal/contract"
)

func TestGraphPushJSONShape(t *testing.T) {
    p := contract.GraphPush{
        Repo:   "tatara-cli",
        Commit: "abc123",
        Files:  []string{"cmd/root.go"},
        Entities: []contract.Entity{{
            ID: "go:func:m.F", Name: "F", Type: contract.EntityGoFunc,
            FilePath: "cmd/root.go", Properties: map[string]string{"resolution": contract.ResTypeResolved},
        }},
        Edges: []contract.Edge{{
            From: "go:func:m.F", To: "go:func:m.G", Relation: contract.RelCalls,
            SrcFile: "cmd/root.go", Properties: map[string]string{"confidence": "0.98"},
        }},
    }
    b, err := json.Marshal(p)
    require.NoError(t, err)
    var got map[string]any
    require.NoError(t, json.Unmarshal(b, &got))
    require.ElementsMatch(t, []string{"repo", "commit", "files", "entities", "edges"}, keys(got))
    ent := got["entities"].([]any)[0].(map[string]any)
    require.ElementsMatch(t, []string{"id", "name", "type", "file_path", "properties"}, keys(ent))
    edge := got["edges"].([]any)[0].(map[string]any)
    require.ElementsMatch(t, []string{"from", "to", "relation", "src_file", "properties"}, keys(edge))
}

func TestIngestItemJSONShape(t *testing.T) {
    it := contract.IngestItem{IdempotencyKey: "k", Text: "t", Metadata: map[string]string{"repo": "r"}}
    b, err := json.Marshal(it)
    require.NoError(t, err)
    var got map[string]any
    require.NoError(t, json.Unmarshal(b, &got))
    require.ElementsMatch(t, []string{"idempotency_key", "text", "metadata"}, keys(got))
}

func keys(m map[string]any) []string {
    out := make([]string, 0, len(m))
    for k := range m {
        out = append(out, k)
    }
    return out
}

// Package contract holds the wire types and vocabulary shared with
// tatara-memory's codegraph and memory APIs. These types mirror the server's
// JSON shapes byte-for-byte; contract_shape_test.go guards against drift.
package contract

// Entity is a node in the code graph.
type Entity struct {
    ID          string            `json:"id"`
    Name        string            `json:"name"`
    Type        string            `json:"type"`
    Description string            `json:"description,omitempty"`
    FilePath    string            `json:"file_path"`
    Properties  map[string]string `json:"properties,omitempty"`
}

// Edge is a directed, typed relationship between two entities.
type Edge struct {
    From       string            `json:"from"`
    To         string            `json:"to"`
    Relation   string            `json:"relation"`
    SrcFile    string            `json:"src_file"`
    Properties map[string]string `json:"properties,omitempty"`
}

// GraphPush is one /code-graph:bulk request.
type GraphPush struct {
    Repo     string   `json:"repo"`
    Commit   string   `json:"commit,omitempty"`
    Files    []string `json:"files"`
    Entities []Entity `json:"entities"`
    Edges    []Edge   `json:"edges"`
}

// PushResult is the /code-graph:bulk response.
type PushResult struct {
    Repo             string `json:"repo"`
    Files            int    `json:"files"`
    EntitiesUpserted int    `json:"entities_upserted"`
    EdgesUpserted    int    `json:"edges_upserted"`
}

// IngestItem is one /memories:bulk item.
type IngestItem struct {
    IdempotencyKey string            `json:"idempotency_key"`
    Text           string            `json:"text"`
    Metadata       map[string]string `json:"metadata,omitempty"`
}

// IngestJob is the /memories:bulk and /ingest-jobs/{id} response.
type IngestJob struct {
    ID     string `json:"id"`
    Status string `json:"status"`
    Total  int    `json:"total"`
    Done   int    `json:"done"`
    Failed int    `json:"failed"`
    Errors []struct {
        IdempotencyKey string `json:"idempotency_key"`
        Error          string `json:"error"`
    } `json:"errors,omitempty"`
}

// Terminal reports whether a job status is final.
func (j IngestJob) Terminal() bool {
    switch j.Status {
    case JobSucceeded, JobFailed, JobPartial:
        return true
    }
    return false
}

// Chunk is an analyzer's semantic-chunk output (assembled into an IngestItem by push).
type Chunk struct {
    EntityID string
    Type     string
    FilePath string
    Language string
    Header   string
    Body     string
}

// Job status values.
const (
    JobSucceeded = "succeeded"
    JobFailed    = "failed"
    JobPartial   = "partial"
)

// Entity types.
const (
    EntityRepo        = "repo"
    EntityFile        = "file"
    EntityGoPackage   = "go_package"
    EntityGoType      = "go_type"
    EntityGoFunc      = "go_func"
    EntityGoMethod    = "go_method"
    EntityPyModule    = "py_module"
    EntityPyClass     = "py_class"
    EntityPyFunc      = "py_func"
    EntityJSModule    = "js_module"
    EntityJSClass     = "js_class"
    EntityJSFunc      = "js_func"
    EntityTFResource  = "tf_resource"
    EntityTFData      = "tf_data"
    EntityTFModule    = "tf_module"
    EntityTFVariable  = "tf_variable"
    EntityTFOutput    = "tf_output"
    EntityHelmChart   = "helm_chart"
    EntityHelmTemplate = "helm_template"
    EntityHelmValue   = "helm_value"
)

// Edge relations.
const (
    RelContains     = "contains"
    RelDefines      = "defines"
    RelImports      = "imports"
    RelCalls        = "calls"
    RelReferences   = "references"
    RelImplements   = "implements"
    RelDependsOn    = "depends_on"
    RelModuleSource = "module_source"
    RelVarRef       = "var_ref"
    RelOutputRef    = "output_ref"
    RelValueRef     = "value_ref"
    RelIncludes     = "includes"
    RelSubchart     = "subchart"
)

// M3 call-edge resolution levels (property key "resolution").
const (
    ResTypeResolved      = "type_resolved"
    ResScopedNameMatch   = "scoped_name_match"
    ResImportedNameMatch = "imported_name_match"
    ResGlobalNameMatch   = "global_name_match"
    ResAmbiguousMultiDef = "ambiguous_multi_def"
    ResUnresolved        = "unresolved"
)

// ConfidenceFor returns the prior confidence string for a resolution level.
func ConfidenceFor(resolution string) string {
    switch resolution {
    case ResTypeResolved:
        return "0.98"
    case ResScopedNameMatch:
        return "0.85"
    case ResImportedNameMatch:
        return "0.7"
    case ResGlobalNameMatch:
        return "0.45"
    case ResAmbiguousMultiDef:
        return "0.2"
    default:
        return "0.0"
    }
}

package config_test

import (
    "testing"
    "time"

    "github.com/stretchr/testify/require"

    "github.com/szymonrychu/tatara-memory-repo-ingester/internal/config"
)

func TestLoadFromEnv(t *testing.T) {
    env := map[string]string{
        "base-url":          "https://memory.example/",
        "oidc-issuer":       "https://auth.example/realms/master",
        "oidc-client-id":    "ingester",
        "oidc-client-secret": "s3cret",
        "oidc-audience":     "tatara-memory",
        "poll-interval":     "2s",
        "http-timeout":      "30s",
    }
    c, err := config.Load(func(k string) string { return env[k] })
    require.NoError(t, err)
    require.Equal(t, "https://memory.example", c.BaseURL) // trailing slash trimmed
    require.Equal(t, "ingester", c.OIDCClientID)
    require.Equal(t, 2*time.Second, c.PollInterval)
    require.Equal(t, 30*time.Second, c.HTTPTimeout)
}

func TestLoadDefaults(t *testing.T) {
    c, err := config.Load(func(string) string { return "" })
    require.NoError(t, err)
    require.Equal(t, 2*time.Second, c.PollInterval)
    require.Equal(t, 60*time.Second, c.HTTPTimeout)
}

// Package config loads ingester configuration from the environment.
package config

import (
    "fmt"
    "strings"
    "time"
)

// Config holds runtime configuration for the ingester.
type Config struct {
    BaseURL          string
    OIDCIssuer       string
    OIDCClientID     string
    OIDCClientSecret string
    OIDCAudience     string
    PollInterval     time.Duration
    HTTPTimeout      time.Duration
}

// Load builds a Config from a key lookup function (kebab-case keys).
func Load(getenv func(string) string) (Config, error) {
    c := Config{
        BaseURL:          strings.TrimRight(getenv("base-url"), "/"),
        OIDCIssuer:       getenv("oidc-issuer"),
        OIDCClientID:     getenv("oidc-client-id"),
        OIDCClientSecret: getenv("oidc-client-secret"),
        OIDCAudience:     getenv("oidc-audience"),
        PollInterval:     2 * time.Second,
        HTTPTimeout:      60 * time.Second,
    }
    if v := getenv("poll-interval"); v != "" {
        d, err := time.ParseDuration(v)
        if err != nil {
            return Config{}, fmt.Errorf("parse poll-interval: %w", err)
        }
        c.PollInterval = d
    }
    if v := getenv("http-timeout"); v != "" {
        d, err := time.ParseDuration(v)
        if err != nil {
            return Config{}, fmt.Errorf("parse http-timeout: %w", err)
        }
        c.HTTPTimeout = d
    }
    return c, nil
}

package analyze_test

import (
    "context"
    "testing"

    "github.com/stretchr/testify/require"

    "github.com/szymonrychu/tatara-memory-repo-ingester/internal/analyze"
    "github.com/szymonrychu/tatara-memory-repo-ingester/internal/contract"
)

type fakeAnalyzer struct {
    name  string
    match func(string) bool
}

func (f fakeAnalyzer) Name() string         { return f.name }
func (f fakeAnalyzer) Match(p string) bool  { return f.match(p) }
func (f fakeAnalyzer) Analyze(_ context.Context, _ string, files []string) (analyze.Result, error) {
    return analyze.Result{Entities: []contract.Entity{{ID: f.name, Name: f.name}}}, nil
}

func TestRegistryGroupsByFirstMatch(t *testing.T) {
    reg := analyze.NewRegistry()
    reg.Register(fakeAnalyzer{name: "go", match: func(p string) bool { return p == "a.go" }})
    reg.Register(fakeAnalyzer{name: "docs", match: func(string) bool { return true }}) // catch-all, lower precedence

    groups := reg.Group([]string{"a.go", "README.md"})
    require.Equal(t, []string{"a.go"}, groups["go"])
    require.Equal(t, []string{"README.md"}, groups["docs"])
}

func TestRegistryUnmatchedFileDropped(t *testing.T) {
    reg := analyze.NewRegistry()
    reg.Register(fakeAnalyzer{name: "go", match: func(p string) bool { return p == "a.go" }})
    groups := reg.Group([]string{"a.go", "weird.xyz"})
    require.Equal(t, map[string][]string{"go": {"a.go"}}, groups)
}

// Package analyze defines the language-neutral analyzer contract and registry.
// Adding a language means adding one file implementing Analyzer and registering
// it; nothing else in the ingester changes.
package analyze

import (
    "context"

    "github.com/szymonrychu/tatara-memory-repo-ingester/internal/contract"
)

// Result is what an Analyzer emits for its assigned file set.
type Result struct {
    Entities []contract.Entity
    Edges    []contract.Edge
    Chunks   []contract.Chunk
}

// Analyzer extracts a code graph and chunks for one language/file class.
type Analyzer interface {
    Name() string
    Match(path string) bool
    Analyze(ctx context.Context, repoRoot string, files []string) (Result, error)
}

// Registry is an ordered set of analyzers; earlier registration wins on Match.
type Registry struct {
    analyzers []Analyzer
}

// NewRegistry returns an empty registry.
func NewRegistry() *Registry { return &Registry{} }

// Register appends an analyzer (precedence = registration order).
func (r *Registry) Register(a Analyzer) { r.analyzers = append(r.analyzers, a) }

// Analyzers returns the registered analyzers in order.
func (r *Registry) Analyzers() []Analyzer { return r.analyzers }

// Group assigns each file to the first analyzer whose Match returns true.
// Files matched by no analyzer are dropped.
func (r *Registry) Group(files []string) map[string][]string {
    groups := map[string][]string{}
    for _, f := range files {
        for _, a := range r.analyzers {
            if a.Match(f) {
                groups[a.Name()] = append(groups[a.Name()], f)
                break
            }
        }
    }
    return groups
}

package walk_test

import (
    "os"
    "os/exec"
    "path/filepath"
    "testing"

    "github.com/stretchr/testify/require"

    "github.com/szymonrychu/tatara-memory-repo-ingester/internal/walk"
)

func gitRepo(t *testing.T) string {
    t.Helper()
    dir := t.TempDir()
    for _, args := range [][]string{
        {"init", "-q"}, {"config", "user.email", "t@t"}, {"config", "user.name", "t"},
    } {
        cmd := exec.Command("git", args...)
        cmd.Dir = dir
        require.NoError(t, cmd.Run())
    }
    return dir
}

func write(t *testing.T, dir, rel, content string) {
    t.Helper()
    p := filepath.Join(dir, rel)
    require.NoError(t, os.MkdirAll(filepath.Dir(p), 0o755))
    require.NoError(t, os.WriteFile(p, []byte(content), 0o644))
}

func commit(t *testing.T, dir, msg string) string {
    t.Helper()
    for _, args := range [][]string{{"add", "-A"}, {"commit", "-q", "-m", msg}} {
        cmd := exec.Command("git", args...)
        cmd.Dir = dir
        require.NoError(t, cmd.Run())
    }
    out, err := exec.Command("git", "-C", dir, "rev-parse", "HEAD").Output()
    require.NoError(t, err)
    return string(out[:len(out)-1])
}

func TestFullWalkListsTrackedFiles(t *testing.T) {
    dir := gitRepo(t)
    write(t, dir, "a.go", "package a")
    write(t, dir, "sub/b.py", "x = 1")
    commit(t, dir, "init")
    files, err := walk.Changed(dir, "", false)
    require.NoError(t, err)
    require.ElementsMatch(t, []string{"a.go", "sub/b.py"}, files)
}

func TestSinceWalkListsOnlyChanged(t *testing.T) {
    dir := gitRepo(t)
    write(t, dir, "a.go", "package a")
    base := commit(t, dir, "init")
    write(t, dir, "c.go", "package c")
    commit(t, dir, "add c")
    files, err := walk.Changed(dir, base, false)
    require.NoError(t, err)
    require.Equal(t, []string{"c.go"}, files)
}

func TestFullFlagOverridesSince(t *testing.T) {
    dir := gitRepo(t)
    write(t, dir, "a.go", "package a")
    base := commit(t, dir, "init")
    write(t, dir, "c.go", "package c")
    commit(t, dir, "add c")
    files, err := walk.Changed(dir, base, true)
    require.NoError(t, err)
    require.ElementsMatch(t, []string{"a.go", "c.go"}, files)
}

// Package walk lists the repository files to ingest.
package walk

import (
    "fmt"
    "os/exec"
    "sort"
    "strings"
)

// Changed returns repo-relative paths to ingest. With full or an empty since, it
// lists all tracked files; otherwise it diffs since..HEAD.
func Changed(repoRoot, since string, full bool) ([]string, error) {
    var args []string
    if full || since == "" {
        args = []string{"-C", repoRoot, "ls-files"}
    } else {
        args = []string{"-C", repoRoot, "diff", "--name-only", since + "..HEAD"}
    }
    out, err := exec.Command("git", args...).Output()
    if err != nil {
        return nil, fmt.Errorf("git %s: %w", strings.Join(args, " "), err)
    }
    var files []string
    for _, line := range strings.Split(strings.TrimSpace(string(out)), "\n") {
        if line != "" {
            files = append(files, line)
        }
    }
    sort.Strings(files)
    return files, nil
}

package push_test

import (
    "context"
    "encoding/json"
    "net/http"
    "net/http/httptest"
    "strings"
    "testing"
    "time"

    "github.com/stretchr/testify/require"

    "github.com/szymonrychu/tatara-memory-repo-ingester/internal/contract"
    "github.com/szymonrychu/tatara-memory-repo-ingester/internal/push"
)

func TestPushGraph(t *testing.T) {
    srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
        require.Equal(t, "/code-graph:bulk", r.URL.Path)
        var p contract.GraphPush
        require.NoError(t, json.NewDecoder(r.Body).Decode(&p))
        require.Equal(t, "tatara-cli", p.Repo)
        w.WriteHeader(200)
        _ = json.NewEncoder(w).Encode(contract.PushResult{Repo: p.Repo, EntitiesUpserted: len(p.Entities)})
    }))
    defer srv.Close()
    c := push.New(srv.URL, http.DefaultClient, time.Millisecond)
    res, err := c.PushGraph(context.Background(), contract.GraphPush{Repo: "tatara-cli", Entities: []contract.Entity{{ID: "x"}}})
    require.NoError(t, err)
    require.Equal(t, 1, res.EntitiesUpserted)
}

func TestPushChunksPollsToTerminal(t *testing.T) {
    var polls int
    srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
        switch {
        case r.URL.Path == "/memories:bulk":
            var req struct{ Items []contract.IngestItem `json:"items"` }
            require.NoError(t, json.NewDecoder(r.Body).Decode(&req))
            require.Len(t, req.Items, 1)
            w.WriteHeader(202)
            _ = json.NewEncoder(w).Encode(contract.IngestJob{ID: "j1", Status: "running"})
        case strings.HasPrefix(r.URL.Path, "/ingest-jobs/"):
            polls++
            st := "running"
            if polls >= 2 {
                st = contract.JobSucceeded
            }
            _ = json.NewEncoder(w).Encode(contract.IngestJob{ID: "j1", Status: st, Total: 1, Done: 1})
        }
    }))
    defer srv.Close()
    c := push.New(srv.URL, http.DefaultClient, time.Millisecond)
    err := c.PushChunks(context.Background(), []contract.IngestItem{{IdempotencyKey: "k", Text: "t"}})
    require.NoError(t, err)
    require.GreaterOrEqual(t, polls, 2)
}

func TestPushChunksPartialIsError(t *testing.T) {
    srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
        if r.URL.Path == "/memories:bulk" {
            w.WriteHeader(202)
            _ = json.NewEncoder(w).Encode(contract.IngestJob{ID: "j1", Status: "running"})
            return
        }
        _ = json.NewEncoder(w).Encode(contract.IngestJob{ID: "j1", Status: contract.JobPartial, Failed: 1})
    }))
    defer srv.Close()
    c := push.New(srv.URL, http.DefaultClient, time.Millisecond)
    err := c.PushChunks(context.Background(), []contract.IngestItem{{IdempotencyKey: "k", Text: "t"}})
    require.Error(t, err)
}

// Package push sends the code graph and semantic chunks to tatara-memory.
package push

import (
    "bytes"
    "context"
    "encoding/json"
    "fmt"
    "io"
    "net/http"
    "time"

    "github.com/szymonrychu/tatara-memory-repo-ingester/internal/contract"
)

// Client posts to a tatara-memory base URL.
type Client struct {
    base         string
    http         *http.Client
    pollInterval time.Duration
}

// New constructs a push client.
func New(base string, hc *http.Client, pollInterval time.Duration) *Client {
    return &Client{base: base, http: hc, pollInterval: pollInterval}
}

// PushGraph posts a GraphPush synchronously and returns the reconciliation summary.
func (c *Client) PushGraph(ctx context.Context, p contract.GraphPush) (contract.PushResult, error) {
    var res contract.PushResult
    if err := c.do(ctx, http.MethodPost, "/code-graph:bulk", p, http.StatusOK, &res); err != nil {
        return contract.PushResult{}, err
    }
    return res, nil
}

// PushChunks posts chunks and polls the resulting job to a terminal state.
func (c *Client) PushChunks(ctx context.Context, items []contract.IngestItem) error {
    if len(items) == 0 {
        return nil
    }
    var job contract.IngestJob
    body := struct {
        Items []contract.IngestItem `json:"items"`
    }{Items: items}
    if err := c.do(ctx, http.MethodPost, "/memories:bulk", body, http.StatusAccepted, &job); err != nil {
        return err
    }
    for !job.Terminal() {
        select {
        case <-ctx.Done():
            return ctx.Err()
        case <-time.After(c.pollInterval):
        }
        if err := c.do(ctx, http.MethodGet, "/ingest-jobs/"+job.ID, nil, http.StatusOK, &job); err != nil {
            return err
        }
    }
    if job.Status != contract.JobSucceeded {
        return fmt.Errorf("ingest job %s ended %s (failed=%d)", job.ID, job.Status, job.Failed)
    }
    return nil
}

func (c *Client) do(ctx context.Context, method, path string, in any, want int, out any) error {
    var rdr io.Reader
    if in != nil {
        b, err := json.Marshal(in)
        if err != nil {
            return fmt.Errorf("marshal %s: %w", path, err)
        }
        rdr = bytes.NewReader(b)
    }
    req, err := http.NewRequestWithContext(ctx, method, c.base+path, rdr)
    if err != nil {
        return fmt.Errorf("request %s: %w", path, err)
    }
    if in != nil {
        req.Header.Set("Content-Type", "application/json")
    }
    resp, err := c.http.Do(req)
    if err != nil {
        return fmt.Errorf("call %s: %w", path, err)
    }
    defer resp.Body.Close()
    if resp.StatusCode != want {
        b, _ := io.ReadAll(io.LimitReader(resp.Body, 2048))
        return fmt.Errorf("%s: status %d: %s", path, resp.StatusCode, string(b))
    }
    if out != nil {
        if err := json.NewDecoder(resp.Body).Decode(out); err != nil {
            return fmt.Errorf("decode %s: %w", path, err)
        }
    }
    return nil
}

package push_test

import (
    "testing"

    "github.com/stretchr/testify/require"

    "github.com/szymonrychu/tatara-memory-repo-ingester/internal/contract"
    "github.com/szymonrychu/tatara-memory-repo-ingester/internal/push"
)

func TestItemsFromChunks(t *testing.T) {
    chunks := []contract.Chunk{{
        EntityID: "go:func:m.F", Type: contract.EntityGoFunc, FilePath: "m.go",
        Language: "go", Header: "[go_func] m.F", Body: "func F() {}",
    }}
    items := push.ItemsFromChunks("tatara-cli", chunks)
    require.Len(t, items, 1)
    require.Equal(t, "[go_func] m.F\n---\nfunc F() {}", items[0].Text)
    require.Equal(t, "go:func:m.F", items[0].Metadata["entity_id"])
    require.Equal(t, "tatara-cli", items[0].Metadata["repo"])
    require.True(t, strings.HasPrefix(items[0].IdempotencyKey, "tatara-cli:go:func:m.F:"))
}

package push

import (
    "crypto/sha256"
    "encoding/hex"
    "fmt"

    "github.com/szymonrychu/tatara-memory-repo-ingester/internal/contract"
)

// ItemsFromChunks turns analyzer chunks into /memories:bulk items.
func ItemsFromChunks(repo string, chunks []contract.Chunk) []contract.IngestItem {
    items := make([]contract.IngestItem, 0, len(chunks))
    for _, ch := range chunks {
        text := ch.Header + "\n---\n" + ch.Body
        sum := sha256.Sum256([]byte(text))
        items = append(items, contract.IngestItem{
            IdempotencyKey: fmt.Sprintf("%s:%s:%s", repo, ch.EntityID, hex.EncodeToString(sum[:8])),
            Text:           text,
            Metadata: map[string]string{
                "repo": repo, "entity_id": ch.EntityID, "type": ch.Type,
                "file_path": ch.FilePath, "language": ch.Language,
            },
        })
    }
    return items
}