goclaw/internal/store/pg/knowledge_graph_temporal.go

package pg

import (
	"context"
	"encoding/json"
	"fmt"
	"time"

	"github.com/nextlevelbuilder/goclaw/internal/store"
)

// ListEntitiesTemporal queries entities with temporal awareness.
// AsOf=nil: current facts only (valid_until IS NULL). AsOf set: facts valid at that time.
func (s *PGKnowledgeGraphStore) ListEntitiesTemporal(ctx context.Context, agentID, userID string, opts store.EntityListOptions, temporal store.TemporalQueryOptions) ([]store.Entity, error) {
	aid := parseUUIDOrNil(agentID)
	limit := opts.Limit
	if limit <= 0 {
		limit = 100
	}

	q := `SELECT id, agent_id, user_id, external_id, name, entity_type, description,
	             properties, source_id, confidence, created_at, updated_at, valid_from, valid_until
	      FROM kg_entities WHERE agent_id = $1 AND user_id = $2`
	args := []any{aid, userID}
	argN := 3

	if opts.EntityType != "" {
		q += fmt.Sprintf(` AND entity_type = $%d`, argN)
		args = append(args, opts.EntityType)
		argN++
	}

	// Temporal filter
	if !temporal.IncludeExpired {
		if temporal.AsOf != nil {
			q += fmt.Sprintf(` AND valid_from <= $%d AND (valid_until IS NULL OR valid_until >= $%d)`, argN, argN)
			args = append(args, *temporal.AsOf)
			argN++
		} else {
			q += ` AND valid_until IS NULL`
		}
	}

	// Tenant scope
	tc, tcArgs, _, err := scopeClause(ctx, argN)
	if err != nil {
		return nil, err
	}
	if tc != "" {
		q += tc
		args = append(args, tcArgs...)
		argN += len(tcArgs)
	}

	q += fmt.Sprintf(` ORDER BY created_at DESC LIMIT $%d OFFSET $%d`, argN, argN+1)
	args = append(args, limit, opts.Offset)

	var tRows []entityTemporalRow
	if err := pkgSqlxDB.SelectContext(ctx, &tRows, q, args...); err != nil {
		return nil, fmt.Errorf("list entities temporal: %w", err)
	}
	entities := make([]store.Entity, len(tRows))
	for i := range tRows {
		entities[i] = tRows[i].toEntity()
	}
	return entities, nil
}

// SupersedeEntity atomically expires the old entity and inserts a replacement.
func (s *PGKnowledgeGraphStore) SupersedeEntity(ctx context.Context, old *store.Entity, replacement *store.Entity) error {
	aid, err := parseUUID(old.AgentID)
	if err != nil {
		return fmt.Errorf("kg supersede entity: %w", err)
	}
	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return fmt.Errorf("supersede begin tx: %w", err)
	}
	defer tx.Rollback()

	now := time.Now().UTC()
	tid := tenantIDForInsert(ctx)

	// Tenant scope for UPDATE
	tc, tcArgs, _, err := scopeClause(ctx, 6)
	if err != nil {
		return err
	}

	// Expire old entity
	expireArgs := append([]any{now, now, aid, old.UserID, old.ExternalID}, tcArgs...)
	_, err = tx.ExecContext(ctx, `
		UPDATE kg_entities SET valid_until = $1, updated_at = $2
		WHERE agent_id = $3 AND user_id = $4 AND external_id = $5 AND valid_until IS NULL`+tc,
		expireArgs...)
	if err != nil {
		return fmt.Errorf("supersede expire old: %w", err)
	}

	// Insert replacement with valid_from = now
	props, _ := json.Marshal(replacement.Properties)
	_, err = tx.ExecContext(ctx, `
		INSERT INTO kg_entities (id, agent_id, user_id, external_id, name, entity_type,
		    description, properties, source_id, confidence, tenant_id, created_at, updated_at, valid_from)
		VALUES (gen_random_uuid(), $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $11, $12)`,
		aid, replacement.UserID, replacement.ExternalID,
		replacement.Name, replacement.EntityType, replacement.Description,
		props, replacement.SourceID, replacement.Confidence, tid, now, now)
	if err != nil {
		return fmt.Errorf("supersede insert new: %w", err)
	}

	return tx.Commit()
}