| 1 | package replication |
| 2 | |
| 3 | import ( |
| 4 | "bufio" |
| 5 | "bytes" |
| 6 | "context" |
| 7 | "encoding/json" |
| 8 | "errors" |
| 9 | "fmt" |
| 10 | "io" |
| 11 | "os" |
| 12 | "path/filepath" |
| 13 | |
| 14 | "github.com/FelineStateMachine/tinyrelay/internal/event" |
| 15 | "github.com/FelineStateMachine/tinyrelay/internal/storage" |
| 16 | ) |
| 17 | |
| 18 | // writeArtifact publishes a complete, durable file by same-directory rename. |
| 19 | // Failed generation leaves any prior artifact intact. |
| 20 | func 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 err |
| 24 | } |
| 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 err |
| 33 | } |
| 34 | if err := os.Rename(file.Name(), path); err != nil { |
| 35 | return err |
| 36 | } |
| 37 | dir, err := os.Open(filepath.Dir(path)) |
| 38 | if err != nil { |
| 39 | return err |
| 40 | } |
| 41 | return errors.Join(dir.Sync(), dir.Close()) |
| 42 | } |
| 43 | |
| 44 | func 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 nil |
| 49 | } |
| 50 | return err |
| 51 | }) |
| 52 | } |
| 53 | |
| 54 | func 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 err |
| 59 | } |
| 60 | line, readErr := reader.ReadBytes('\n' ) |
| 61 | if len(bytes.TrimSpace(line)) > 0 { |
| 62 | var item event.Event |
| 63 | 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 nil |
| 75 | } |
| 76 | if readErr != nil { |
| 77 | return readErr |
| 78 | } |
| 79 | } |
| 80 | } |
| 81 | |
| 82 | func 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 err |
| 86 | } |
| 87 | encoder := json.NewEncoder(output) |
| 88 | for _, row := range rows { |
| 89 | if err := ctx.Err(); err != nil { |
| 90 | return err |
| 91 | } |
| 92 | if err := encoder.Encode(row.Event); err != nil { |
| 93 | return fmt.Errorf("replication: write dump: %w" , err) |
| 94 | } |
| 95 | } |
| 96 | return nil |
| 97 | } |