package runtime

import (
	"bufio"
	"context"
	"fmt"
	"io"
	"net/url"
	"os"
	"strings"

	"browser.local/run/protocol"
)

type LogSourceCheckpoint struct {
	SourceKey string
	Offset    int64
	Sequence  uint64
	CursorRef string
}

type LogCheckpointStore interface {
	GetLogCheckpoint(sourceKey string) LogSourceCheckpoint
	PutLogCheckpoint(checkpoint LogSourceCheckpoint)
}

type MemoryLogCheckpointStore struct {
	checkpoints map[string]LogSourceCheckpoint
}

func NewMemoryLogCheckpointStore() *MemoryLogCheckpointStore {
	return &MemoryLogCheckpointStore{checkpoints: map[string]LogSourceCheckpoint{}}
}

func (store *MemoryLogCheckpointStore) GetLogCheckpoint(sourceKey string) LogSourceCheckpoint {
	if store == nil || store.checkpoints == nil {
		return LogSourceCheckpoint{SourceKey: sourceKey}
	}
	return store.checkpoints[sourceKey]
}

func (store *MemoryLogCheckpointStore) PutLogCheckpoint(checkpoint LogSourceCheckpoint) {
	if store == nil {
		return
	}
	if store.checkpoints == nil {
		store.checkpoints = map[string]LogSourceCheckpoint{}
	}
	store.checkpoints[checkpoint.SourceKey] = checkpoint
}

func TailDeclaredFileLogSource(ctx context.Context, workspaceRoot string, assignment protocol.RunJobAssignment, source RuntimeLogSource, sink ProcessLogSink, store LogCheckpointStore) LifecycleExecutionResult {
	if source.Kind != "file.tail" {
		return lifecycleFailure("unsupported_log_source", "only file.tail sources are supported by the local tailer")
	}
	if !protocol.ValidLogicalFileKey(source.Key) || !protocol.ValidLogicalFileKey(source.TargetKey) || !protocol.ValidLogicalFileKey(source.StreamKey) {
		return lifecycleFailure("unsafe_log_source", "log source is unsafe")
	}
	if sink == nil {
		sink = NoopProcessLogSink{}
	}
	if store == nil {
		store = NewMemoryLogCheckpointStore()
	}
	serverRoot, err := scopedServerWorkspace(workspaceRoot, assignment.ServerInstanceID)
	if err != nil {
		return lifecycleFailure("unsafe_log_workspace", err.Error())
	}
	path, err := scopedPath(serverRoot, source.TargetKey)
	if err != nil {
		return lifecycleFailure("unsafe_log_source", err.Error())
	}
	file, err := os.Open(path)
	if err != nil {
		return lifecycleFailure("log_source_open_failed", err.Error())
	}
	defer file.Close()

	checkpoint := store.GetLogCheckpoint(source.Key)
	if checkpoint.Offset > 0 {
		if _, err := file.Seek(checkpoint.Offset, 0); err != nil {
			return lifecycleFailure("log_source_seek_failed", err.Error())
		}
	}
	reader := bufio.NewReader(file)
	for {
		// A newline only frames an entry. Every other byte, including CR, blank
		// lines, and arbitrarily long output, remains untouched.
		line, readErr := reader.ReadString('\n')
		if len(line) > 0 {
			checkpoint.Sequence++
			if err := sink.Append(ctx, assignment, source.StreamKey, strings.TrimSuffix(line, "\n")); err != nil {
				return lifecycleFailure("log_source_sink_failed", err.Error())
			}
			checkpoint.SourceKey = source.Key
			checkpoint.Offset += int64(len(line))
			checkpoint.CursorRef = fmt.Sprintf("artifact://jobs/%s/live-log-checkpoint", url.PathEscape(assignment.JobID))
			store.PutLogCheckpoint(checkpoint)
		}
		if readErr == nil {
			continue
		}
		if readErr == io.EOF {
			break
		}
		return lifecycleFailure("log_source_read_failed", readErr.Error())
	}
	if checkpoint.CursorRef == "" {
		checkpoint.CursorRef = fmt.Sprintf("artifact://jobs/%s/live-log-checkpoint", url.PathEscape(assignment.JobID))
	}
	return LifecycleExecutionResult{
		State:     lifecycleResultStateSucceeded,
		Progress:  protocol.RunJobProgressReport{Percent: 100, Message: "live log checkpoint updated"},
		ResultRef: checkpoint.CursorRef,
		Message:   "live log source tailed with durable offset checkpoint",
	}
}

func RedactedLogCheckpointSummary(checkpoint LogSourceCheckpoint) string {
	return strings.Join([]string{
		"source=" + checkpoint.SourceKey,
		fmt.Sprintf("offset=%d", checkpoint.Offset),
		fmt.Sprintf("sequence=%d", checkpoint.Sequence),
		"cursorRef=" + checkpoint.CursorRef,
	}, " ")
}