hyperguild/ingestion/internal/graphsync/graphsync.go

// Package graphsync glues the disk-resident brain markdown documents to
// the relational graph in [graphstore]. It is a tiny seam so that the
// MCP handlers can call one function after every successful write or
// ingest without having to know either the parser or the postgres
// schema.
//
// Every operation is best-effort from the caller's perspective: if the
// graph store is unconfigured or the doc parses to nothing usable, the
// helpers return nil. Real database errors are surfaced so the caller
// can log them.
package graphsync

import (
	"context"
	"errors"
	"fmt"
	"io/fs"
	"path"
	"path/filepath"
	"strings"

	"github.com/mathiasbq/hyperguild/ingestion/internal/brainfs"
	"github.com/mathiasbq/hyperguild/ingestion/internal/graph"
	"github.com/mathiasbq/hyperguild/ingestion/internal/graphstore"
)

// Store is the subset of graphstore.PGStore that graphsync requires.
// Tests can substitute a fake by satisfying this interface.
type Store interface {
	UpsertEntity(ctx context.Context, e graph.Entity) error
	ReplaceEdgesForDoc(ctx context.Context, docPath string, edges []graph.Edge) error
	DeleteByDoc(ctx context.Context, docPath string) error
}

// Compile-time assertion that *graphstore.PGStore satisfies Store.
var _ Store = (*graphstore.PGStore)(nil)

// IndexDoc reads docPath under brainDir and pushes one Entity + its
// outgoing wikilink Edges into store. relPath must be the
// forward-slash path relative to brainDir (the same shape returned by
// api.WriteNote).
//
// nil store is a valid no-op so callers can wire the helper
// unconditionally and let configuration decide whether the graph is
// populated.
func IndexDoc(ctx context.Context, store Store, brainDir, relPath string) error {
	if store == nil {
		return nil
	}
	if relPath == "" {
		return nil
	}
	// Callers pass caller-supplied paths (brain_tunnel, brain_update): index
	// only a clean path under wiki/ or knowledge/.
	if path.Clean(relPath) != relPath || !filepath.IsLocal(relPath) ||
		(!strings.HasPrefix(relPath, "wiki/") && !strings.HasPrefix(relPath, "knowledge/")) {
		return fmt.Errorf("index %q: not a clean wiki/ or knowledge/ path", relPath)
	}
	bfs, err := brainfs.Open(brainDir)
	if err != nil {
		return fmt.Errorf("open brain: %w", err)
	}
	defer func() { _ = bfs.Close() }()
	return indexDoc(ctx, store, bfs, relPath)
}

// indexDoc reads relPath through the brain root, regular files only
// (hyperguild#157): the brain is a Syncthing folder, and os.ReadFile
// followed a synced symlink into the graph and blocked on a FIFO.
func indexDoc(ctx context.Context, store Store, bfs *brainfs.Brain, relPath string) error {
	content, err := bfs.ReadRegular(relPath)
	if err != nil {
		return fmt.Errorf("read %q: %w", relPath, err)
	}
	ent, edges, ok := graph.Extract(relPath, content)
	if !ok {
		return nil
	}
	if err := store.UpsertEntity(ctx, ent); err != nil {
		return fmt.Errorf("upsert entity: %w", err)
	}
	if err := store.ReplaceEdgesForDoc(ctx, relPath, edges); err != nil {
		return fmt.Errorf("replace edges: %w", err)
	}
	return nil
}

// BackfillFromBrainDir walks every markdown file under brainDir/wiki/
// and brainDir/knowledge/, parses each, and upserts the resulting
// Entity + Edges. Existing rows are overwritten; orphan rows for
// already-deleted files are NOT cleaned up — call this only on a
// fresh store, or follow with a separate prune pass.
//
// Intended for one-shot startup runs against a populated brain dir.
// Cost scales linearly with corpus size; ~30 wiki pages plus the
// knowledge corpus is a few hundred ms.
func BackfillFromBrainDir(ctx context.Context, store Store, brainDir string) (indexed int, _ error) {
	if store == nil {
		return 0, nil
	}
	bfs, err := brainfs.Open(brainDir)
	if err != nil {
		return 0, fmt.Errorf("open brain: %w", err)
	}
	defer func() { _ = bfs.Close() }()
	for _, root := range []string{"wiki", "knowledge"} {
		if _, err := bfs.Lstat(root); errors.Is(err, fs.ErrNotExist) {
			continue
		}
		// Any walk error, including a root that is a symlink, aborts.
		var walkErr error
		err := bfs.WalkRegular(root, func(rel string) error {
			if path.Ext(rel) != ".md" {
				return nil
			}
			if err := indexDoc(ctx, store, bfs, rel); err != nil {
				return fmt.Errorf("index %q: %w", rel, err)
			}
			indexed++
			return nil
		}, func(rel string, err error) {
			if walkErr == nil {
				walkErr = fmt.Errorf("%s: %w", rel, err)
			}
		})
		if err == nil {
			err = walkErr
		}
		if err != nil {
			return indexed, fmt.Errorf("walk %s: %w", root, err)
		}
	}
	return indexed, nil
}