1package daemon23import (4 "bytes"5 "context"6 "encoding/json"7 "os"8 "path/filepath"9 "strings"10 "testing"11 "time"1213 "github.com/FelineStateMachine/tinyrelay/internal/event"14 "github.com/FelineStateMachine/tinyrelay/internal/replication"15 "github.com/FelineStateMachine/tinyrelay/internal/storage"16)1718func TestMountedHTTPPortableImportReportsFinalCountAndContent(t *testing.T) {19 app, tenant := testTenant(t)20 secret := strings.Repeat("0", 63) + "1"21 items := make([]event.Event, 2)22 var body bytes.Buffer23 for i := range items {24 items[i] = event.Event{Kind: 1, CreatedAt: time.Now().Unix() + int64(i), Content: "portable-http-" + string(rune('a'+i))}25 if err := event.Sign(&items[i], secret); err != nil {26 t.Fatal(err)27 }28 raw, err := json.Marshal(items[i])29 if err != nil {30 t.Fatal(err)31 }32 body.Write(raw)33 body.WriteByte('\n')34 }35 res := mountedRequest(t, app, "PUT", "/import", body.Bytes(), "", true)36 if res.Code != 200 {37 t.Fatalf("mounted import status=%d body=%s", res.Code, res.Body.String())38 }39 var response struct {40 Result struct {41 Imported int `json:"imported"`42 } `json:"result"`43 }44 if err := json.Unmarshal(res.Body.Bytes(), &response); err != nil {45 t.Fatal(err)46 }47 if response.Result.Imported != len(items) {48 t.Fatalf("mounted import count=%d want=%d", response.Result.Imported, len(items))49 }50 for _, item := range items {51 rows, err := tenant.store.Query(context.Background(), event.Filter{IDs: []string{item.ID}, Tags: map[string][]string{}}, storage.QueryOptions{Now: time.Now().Unix(), Access: storage.Access{All: true}})52 if err != nil || len(rows.Events) != 1 || rows.Events[0].Content != item.Content {53 t.Fatalf("imported content missing for %s: %d %v", item.ID, len(rows.Events), err)54 }55 }56}5758func TestDaemonPortableImportJobResumesAfterRestart(t *testing.T) {59 ctx := context.Background()60 dir := t.TempDir()61 secret := strings.Repeat("0", 63) + "1"62 owner, err := event.PublicKey(secret)63 if err != nil {64 t.Fatal(err)65 }66 config := Config{DataDir: dir, DefaultTenant: "main"}67 app, err := New(ctx, config)68 if err != nil {69 t.Fatal(err)70 }71 meta, err := app.Create(ctx, CreateOptions{Name: "main", Owner: owner, Template: "default"})72 if err != nil {73 t.Fatal(err)74 }75 if err := app.Start(ctx, "http://relay.test"); err != nil {76 t.Fatal(err)77 }78 tenant := app.tenants[meta.ID]79 tenant.workCancel()80 tenant.workWG.Wait()81 item := event.Event{Kind: 1, CreatedAt: time.Now().Unix(), Content: "portable import"}82 if err := event.Sign(&item, secret); err != nil {83 t.Fatal(err)84 }85 raw, err := json.Marshal(item)86 if err != nil {87 t.Fatal(err)88 }89 if err := os.WriteFile(filepath.Join(meta.Paths.Root, "resume.jsonl"), append(raw, '\n'), 0600); err != nil {90 t.Fatal(err)91 }92 if err := tenant.replication.AddJob(ctx, replication.JobSpec{ID: "resume-import", Kind: replication.JobImport, Relays: []string{"resume.jsonl"}}); err != nil {93 t.Fatal(err)94 }95 if err := app.Close(ctx); err != nil {96 t.Fatal(err)97 }9899 app, err = New(ctx, config)100 if err != nil {101 t.Fatal(err)102 }103 defer app.Close(ctx)104 if err := app.Start(ctx, "http://relay.test"); err != nil {105 t.Fatal(err)106 }107 reloaded := app.tenants[meta.ID]108 if !waitFor(t, 8*time.Second, func() bool {109 statuses, statusErr := reloaded.replication.ListJobStatus(ctx)110 return statusErr == nil && len(statuses) == 1 && statuses[0].Phase == "complete"111 }) {112 statuses, _ := reloaded.replication.ListJobStatus(ctx)113 t.Fatalf("portable import did not resume: %#v", statuses)114 }115 result, err := reloaded.store.Query(ctx, event.Filter{IDs: []string{item.ID}, Tags: map[string][]string{}}, storage.QueryOptions{Now: time.Now().Unix(), Access: storage.Access{All: true}})116 if err != nil || len(result.Events) != 1 {117 t.Fatalf("imported event missing after restart: %d %v", len(result.Events), err)118 }119}