1package replication23import (4 "bufio"5 "bytes"6 "context"7 "encoding/json"8 "errors"9 "fmt"10 "io"11 "os"12 "path/filepath"1314 "github.com/FelineStateMachine/tinyrelay/internal/event"15 "github.com/FelineStateMachine/tinyrelay/internal/storage"16)1718// writeArtifact publishes a complete, durable file by same-directory rename.19// Failed generation leaves any prior artifact intact.20func writeArtifact(path string, produce func(io.Writer) error) error {21 file, err := os.CreateTemp(filepath.Dir(path), ".tiny-artifact-")22 if err != nil {23 return err24 }25 defer os.Remove(file.Name())26 err = produce(file)27 if err == nil {28 err = file.Sync()29 }30 err = errors.Join(err, file.Close())31 if err != nil {32 return err33 }34 if err := os.Rename(file.Name(), path); err != nil {35 return err36 }37 dir, err := os.Open(filepath.Dir(path))38 if err != nil {39 return err40 }41 return errors.Join(dir.Sync(), dir.Close())42}4344func ImportJSONL(ctx context.Context, store *storage.Store, input io.Reader, now int64) error {45 return readJSONLEvents(ctx, input, func(item event.Event) error {46 _, err := store.Save(ctx, item, storage.SaveOptions{Now: now})47 if errors.Is(err, storage.ErrDuplicate) || errors.Is(err, storage.ErrReplaced) {48 return nil49 }50 return err51 })52}5354func readJSONLEvents(ctx context.Context, input io.Reader, accept func(event.Event) error) error {55 reader := bufio.NewReader(input)56 for lineNumber := 1; ; lineNumber++ {57 if err := ctx.Err(); err != nil {58 return err59 }60 line, readErr := reader.ReadBytes('\n')61 if len(bytes.TrimSpace(line)) > 0 {62 var item event.Event63 if err := json.Unmarshal(line, &item); err != nil {64 return fmt.Errorf("import line %d: %w", lineNumber, err)65 }66 if err := event.Validate(item); err != nil {67 return fmt.Errorf("import line %d: %w", lineNumber, err)68 }69 if err := accept(item); err != nil {70 return fmt.Errorf("import line %d: %w", lineNumber, err)71 }72 }73 if errors.Is(readErr, io.EOF) {74 return nil75 }76 if readErr != nil {77 return readErr78 }79 }80}8182func WriteDump(ctx context.Context, store *storage.Store, output io.Writer, now int64) error {83 rows, err := store.After(ctx, 0, event.Filter{Tags: map[string][]string{}}, storage.QueryOptions{Now: now, Access: storage.Access{All: true}, Limit: 0})84 if err != nil {85 return err86 }87 encoder := json.NewEncoder(output)88 for _, row := range rows {89 if err := ctx.Err(); err != nil {90 return err91 }92 if err := encoder.Encode(row.Event); err != nil {93 return fmt.Errorf("replication: write dump: %w", err)94 }95 }96 return nil97}