kenn-io--agentsview/internal/sync/s3.go

package sync

import (
	"context"
	"fmt"
	"io"
	"os"
	"path"
	"path/filepath"
	"strings"

	"go.kenn.io/agentsview/internal/parser"
)

func s3SessionIDPrefix(machine string) string {
	if machine == "" {
		return ""
	}
	return machine + "~"
}

func applyIDPrefixToID(prefix, id string) string {
	if prefix == "" || id == "" || strings.HasPrefix(id, prefix) {
		return id
	}
	return prefix + id
}

func applyIDPrefixToIDs(prefix string, ids []string) []string {
	if prefix == "" || len(ids) == 0 {
		return ids
	}
	prefixed := make([]string, len(ids))
	for i, id := range ids {
		prefixed[i] = applyIDPrefixToID(prefix, id)
	}
	return prefixed
}

func applyIDPrefixToParsedResult(
	prefix string, result *parser.ParseResult,
) {
	if prefix == "" {
		return
	}
	result.Session.ID = applyIDPrefixToID(prefix, result.Session.ID)
	result.Session.ParentSessionID = applyIDPrefixToID(
		prefix, result.Session.ParentSessionID,
	)
	result.Session.SourceSessionID = applyIDPrefixToID(
		prefix, result.Session.SourceSessionID,
	)
	for i := range result.Messages {
		for j := range result.Messages[i].ToolCalls {
			call := &result.Messages[i].ToolCalls[j]
			call.SubagentSessionID = applyIDPrefixToID(
				prefix, call.SubagentSessionID,
			)
			for k := range call.ResultEvents {
				event := &call.ResultEvents[k]
				event.SubagentSessionID = applyIDPrefixToID(
					prefix, event.SubagentSessionID,
				)
			}
		}
	}
}

func safeS3TempRelPath(file parser.DiscoveredFile) (string, error) {
	trimmed := strings.TrimPrefix(file.Path, "s3://")
	parts := strings.Split(trimmed, "/")
	relParts := parts
	if len(parts) > 1 {
		relParts = parts[1:]
	}
	if file.Agent == parser.AgentClaude {
		for i := 0; i+1 < len(parts); i++ {
			if parts[i] == "raw" && parts[i+1] == "claude" {
				relParts = parts[i+2:]
				break
			}
		}
	}
	if file.Agent == parser.AgentCodex {
		for i := 0; i+1 < len(parts); i++ {
			if parts[i] == "raw" && parts[i+1] == "codex" {
				relParts = parts[i+2:]
				break
			}
		}
		relParts = codexS3TempRelParts(relParts)
	}
	if len(relParts) == 0 {
		return "", fmt.Errorf("unsafe s3 object name: %q", file.Path)
	}
	for _, part := range relParts {
		if part == "" || part == "." || part == ".." ||
			strings.ContainsAny(part, `\/`) {
			return "", fmt.Errorf("unsafe s3 object name: %q", file.Path)
		}
	}
	return filepath.Join(relParts...), nil
}

func codexS3TempRelParts(parts []string) []string {
	for i, part := range parts {
		if part == "sessions" || part == "archived_sessions" {
			return parts[i:]
		}
	}
	if len(parts) == 0 {
		return parts
	}
	return append([]string{"sessions"}, parts...)
}

func hydrateS3CodexSessionIndex(sessionPath, sessionURI string) (string, error) {
	indexURI, ok := parser.CodexS3SessionIndexURI(sessionURI)
	if !ok {
		return "", nil
	}
	local := localCodexSessionIndexPath(sessionPath)
	if local == "" {
		return "", nil
	}
	rc, err := fetchS3Object(indexURI)
	if err != nil {
		if isMissingS3Object(err) {
			return "", nil
		}
		return "", err
	}
	defer rc.Close()
	if err := os.MkdirAll(filepath.Dir(local), 0o755); err != nil {
		return "", err
	}
	f, err := os.Create(local)
	if err != nil {
		return "", err
	}
	if _, err := io.Copy(f, rc); err != nil {
		f.Close()
		_ = os.Remove(local)
		if isMissingS3Object(err) {
			return "", nil
		}
		return "", err
	}
	if err := f.Close(); err != nil {
		return "", err
	}
	return local, nil
}

func localCodexSessionIndexPath(sessionPath string) string {
	dir := filepath.Dir(sessionPath)
	for dir != "" {
		base := filepath.Base(dir)
		if base == "sessions" || base == "archived_sessions" {
			return filepath.Join(
				filepath.Dir(dir), parser.CodexSessionIndexFilename,
			)
		}
		parent := filepath.Dir(dir)
		if parent == dir {
			break
		}
		dir = parent
	}
	return ""
}

// processS3Session reads a Claude/Codex session JSONL directly from object
// storage (in-process, no persistent local mirror): download the object's
// bytes, buffer them to a transient temp file so the existing path-based
// parsers (incremental offsets, subagent layout) work unchanged, run the
// normal per-agent processor, then delete the temp file.
func (e *Engine) processS3Session(
	ctx context.Context, file parser.DiscoveredFile, sourceInfo os.FileInfo,
) processResult {
	idPrefix := s3SessionIDPrefix(file.Machine)
	sourceFingerprint := s3SourceFingerprint(file)
	sourceChanged := e.s3SourceMetadataChangedFromInfo(
		file,
		sourceInfo.Size(),
		sourceInfo.ModTime().UnixNano(),
		sourceFingerprint,
	)
	switch file.Agent {
	case parser.AgentClaude:
		sessionID := strings.TrimSuffix(sourceInfo.Name(), ".jsonl")
		fullID := applyIDPrefixToID(idPrefix, sessionID)
		if e.shouldSkipFileWithPrefix(
			idPrefix, sessionID, sourceInfo, sourceFingerprint,
		) &&
			e.db.GetSessionFilePath(fullID) == file.Path {
			sess, _ := e.db.GetSession(ctx, fullID)
			if sess != nil &&
				sess.Project != "" &&
				!parser.NeedsProjectReparse(sess.Project) {
				return processResult{skip: true}
			}
		}
	case parser.AgentCodex:
		if uuid := parser.CodexSessionUUIDFromFilename(
			path.Base(file.Path),
		); uuid != "" {
			sessionID := "codex:" + uuid
			fullID := applyIDPrefixToID(idPrefix, sessionID)
			if e.shouldSkipFileWithPrefix(
				idPrefix, sessionID, sourceInfo, sourceFingerprint,
			) &&
				e.db.GetSessionFilePath(fullID) == file.Path {
				sess, _ := e.db.GetSession(ctx, fullID)
				indexNameChanged := false
				if sess != nil &&
					sess.Project != "" &&
					!parser.NeedsProjectReparse(sess.Project) {
					var indexErr error
					indexNameChanged, indexErr =
						e.s3CodexIndexSessionNameChanged(file, uuid)
					if indexErr != nil {
						return processResult{
							err:         indexErr,
							noCacheSkip: true,
						}
					}
				}
				if sess != nil &&
					sess.Project != "" &&
					!parser.NeedsProjectReparse(sess.Project) &&
					!indexNameChanged {
					return processResult{skip: true}
				}
			}
		}
	}

	relPath, err := safeS3TempRelPath(file)
	if err != nil {
		return processResult{err: err}
	}
	rc, err := fetchS3Object(file.Path)
	if err != nil {
		return processResult{err: err, noCacheSkip: true}
	}
	defer rc.Close()
	dir, err := os.MkdirTemp("", "avs3-")
	if err != nil {
		return processResult{err: err, noCacheSkip: true}
	}
	defer os.RemoveAll(dir)
	tmp := filepath.Join(dir, relPath)
	if err := os.MkdirAll(filepath.Dir(tmp), 0o755); err != nil {
		return processResult{err: err, noCacheSkip: true}
	}
	f, err := os.Create(tmp)
	if err != nil {
		return processResult{err: err, noCacheSkip: true}
	}
	// Stream the object straight to disk so a large session never has to be
	// held whole in memory.
	if _, err := io.Copy(f, rc); err != nil {
		f.Close()
		return processResult{err: err, noCacheSkip: true}
	}
	if err := f.Close(); err != nil {
		return processResult{err: err, noCacheSkip: true}
	}
	hydratedToolResults := false
	sawPersistedToolResults := false
	switch file.Agent {
	case parser.AgentClaude:
		rewrote, sawPersisted, err := hydrateS3ClaudeToolResults(tmp, file.Path)
		if err != nil {
			return processResult{err: err, noCacheSkip: true}
		}
		hydratedToolResults = rewrote
		sawPersistedToolResults = sawPersisted
	case parser.AgentCodex:
		indexPath, err := hydrateS3CodexSessionIndex(tmp, file.Path)
		if err != nil {
			return processResult{err: err, noCacheSkip: true}
		}
		if indexPath != "" {
			defer parser.EvictCodexSessionIndex(indexPath)
		}
	}
	res, err := e.parseMaterializedS3Source(ctx, file, dir, tmp)
	if err != nil {
		return processResult{err: err, noCacheSkip: true}
	}
	// Record the real s3:// source on each parsed session rather than the
	// transient temp path (which is deleted on return), so the stored source
	// pointer reflects where the session actually came from.
	for i := range res.results {
		applyIDPrefixToParsedResult(idPrefix, &res.results[i])
		res.results[i].Session.File.Path = file.Path
		res.results[i].Session.File.Size = sourceInfo.Size()
		res.results[i].Session.File.Mtime = sourceInfo.ModTime().UnixNano()
		if sourceFingerprint != "" {
			res.results[i].Session.File.Hash = sourceFingerprint
		}
	}
	if sourceChanged || hydratedToolResults || sawPersistedToolResults {
		res.forceReplace = true
	}
	res.excludedSessionIDs = applyIDPrefixToIDs(
		idPrefix, res.excludedSessionIDs,
	)
	return res
}

// parseMaterializedS3Source parses an s3:// object that has been materialized to
// a local temp file through the normal provider parse path, instead of a
// per-agent S3 parse method. The temp file is laid out under tempDir at the
// prefix-anchored path the parser expects (so the filename-derived session
// identity and any companion lookups resolve), and the provider is configured
// with tempDir as its root so pathFromSource resolves the temp path. This is the
// same parse a local source of the same agent would get; the only S3-specific
// handling (machine-ID namespacing, recording the s3:// URI as the source path,
// and forced replacement on source change) is applied by the caller.
func (e *Engine) parseMaterializedS3Source(
	ctx context.Context,
	file parser.DiscoveredFile,
	tempDir, tempPath string,
) (processResult, error) {
	machine := e.machine
	if file.Machine != "" {
		machine = file.Machine // the s3 source machine overrides the host
	}
	provider, ok := parser.NewProvider(file.Agent, parser.ProviderConfig{
		Roots:   []string{tempDir},
		Machine: machine,
	})
	if !ok {
		return processResult{}, fmt.Errorf(
			"no provider for s3 agent %s", file.Agent,
		)
	}
	source := parser.SourceRef{
		Provider:       file.Agent,
		Key:            tempPath,
		DisplayPath:    tempPath,
		FingerprintKey: tempPath,
		ProjectHint:    file.Project,
		Opaque:         parser.MaterializedFileSource{Path: tempPath},
	}
	outcome, err := provider.Parse(ctx, parser.ParseRequest{
		Source:     source,
		Machine:    machine,
		ForceParse: true,
	})
	if err != nil {
		return processResult{}, err
	}
	// Do not short-circuit on an empty Results slice: a content-free source can
	// still carry ExcludedSessionIDs (a Claude /usage probe parses to no live
	// session but excludes its ID), and the caller needs those IDs to drop the
	// previously-archived row on resync. ForceReplace must survive too.
	return processResult{
		results:            parseOutcomeResults(outcome.Results),
		excludedSessionIDs: append([]string(nil), outcome.ExcludedSessionIDs...),
		forceReplace:       outcome.ForceReplace,
	}, nil
}