| 1 | package main |
| 2 | |
| 3 | import ( |
| 4 | "encoding/json" |
| 5 | "errors" |
| 6 | "maps" |
| 7 | "os" |
| 8 | "path/filepath" |
| 9 | |
| 10 | "reasonix/internal/config" |
| 11 | "reasonix/internal/fileutil" |
| 12 | "reasonix/internal/identitylock" |
| 13 | ) |
| 14 | |
| 15 | type historicalImportQueueSidecar struct { |
| 16 | Version int `json:"version"` |
| 17 | QueueRevision uint64 `json:"queueRevision,omitempty"` |
| 18 | Current string `json:"current,omitempty"` |
| 19 | Queue []string `json:"queue"` |
| 20 | Presentations map[string]historicalSourcePresentation `json:"presentations,omitempty"` |
| 21 | extra map[string]json.RawMessage |
| 22 | } |
| 23 | |
| 24 | type historicalSourcePresentation struct { |
| 25 | Title string `json:"title,omitempty"` |
| 26 | Pinned *bool `json:"pinned,omitempty"` |
| 27 | extra map[string]json.RawMessage |
| 28 | } |
| 29 | |
| 30 | func (s *historicalImportQueueSidecar) UnmarshalJSON(data []byte) error { |
| 31 | type plain historicalImportQueueSidecar |
| 32 | if err := json.Unmarshal(data, (*plain)(s)); err != nil { |
| 33 | return err |
| 34 | } |
| 35 | return json.Unmarshal(data, &s.extra) |
| 36 | } |
| 37 | |
| 38 | func (s historicalImportQueueSidecar) MarshalJSON() ([]byte, error) { |
| 39 | type plain historicalImportQueueSidecar |
| 40 | return marshalHistoricalFields(plain(s), s.extra, "version", "queueRevision", "current", "queue", "presentations") |
| 41 | } |
| 42 | |
| 43 | func (p *historicalSourcePresentation) UnmarshalJSON(data []byte) error { |
| 44 | type plain historicalSourcePresentation |
| 45 | if err := json.Unmarshal(data, (*plain)(p)); err != nil { |
| 46 | return err |
| 47 | } |
| 48 | return json.Unmarshal(data, &p.extra) |
| 49 | } |
| 50 | |
| 51 | func (p historicalSourcePresentation) MarshalJSON() ([]byte, error) { |
| 52 | type plain historicalSourcePresentation |
| 53 | return marshalHistoricalFields(plain(p), p.extra, "title", "pinned") |
| 54 | } |
| 55 | |
| 56 | func marshalHistoricalFields(value any, extra map[string]json.RawMessage, known ...string) ([]byte, error) { |
| 57 | fields := maps.Clone(extra) |
| 58 | if fields == nil { |
| 59 | fields = map[string]json.RawMessage{} |
| 60 | } |
| 61 | for _, key := range known { |
| 62 | delete(fields, key) |
| 63 | } |
| 64 | data, err := json.Marshal(value) |
| 65 | if err != nil { |
| 66 | return nil, err |
| 67 | } |
| 68 | if err := json.Unmarshal(data, &fields); err != nil { |
| 69 | return nil, err |
| 70 | } |
| 71 | return json.Marshal(fields) |
| 72 | } |
| 73 | |
| 74 | func historicalImportQueuePath() string { |
| 75 | return filepath.Join(filepath.Dir(config.DesktopWorkspaceStatePath()), "historical-import-queue.v1.json") |
| 76 | } |
| 77 | |
| 78 | func readHistoricalSidecar() (historicalImportQueueSidecar, error) { |
| 79 | saved := historicalImportQueueSidecar{Version: 1, Queue: []string{}, Presentations: map[string]historicalSourcePresentation{}} |
| 80 | data, err := os.ReadFile(historicalImportQueuePath()) |
| 81 | if os.IsNotExist(err) { |
| 82 | return saved, nil |
| 83 | } |
| 84 | if err != nil { |
| 85 | return saved, err |
| 86 | } |
| 87 | if err := json.Unmarshal(data, &saved); err != nil { |
| 88 | return saved, err |
| 89 | } |
| 90 | if saved.Version != 1 { |
| 91 | return saved, errors.New("unsupported historical scheduling version") |
| 92 | } |
| 93 | if saved.Presentations == nil { |
| 94 | saved.Presentations = map[string]historicalSourcePresentation{} |
| 95 | } |
| 96 | return saved, nil |
| 97 | } |
| 98 | |
| 99 | // The file lock covers read/modify/write, not just atomic replacement. Queue |
| 100 | // writes preserve current presentation fields; presentation writes never alter a batch. |
| 101 | func updateHistoricalSidecar(mutate func(*historicalImportQueueSidecar) error) error { |
| 102 | path := historicalImportQueuePath() |
| 103 | if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil { |
| 104 | return err |
| 105 | } |
| 106 | release, err := identitylock.TryAcquire(path + ".lock") |
| 107 | if err != nil { |
| 108 | return err |
| 109 | } |
| 110 | defer release() |
| 111 | saved, err := readHistoricalSidecar() |
| 112 | if err != nil { |
| 113 | return err |
| 114 | } |
| 115 | if err := mutate(&saved); err != nil { |
| 116 | return err |
| 117 | } |
| 118 | data, err := json.Marshal(saved) |
| 119 | if err != nil { |
| 120 | return err |
| 121 | } |
| 122 | return fileutil.AtomicWriteFileStrict(path, append(data, '\n'), 0o600) |
| 123 | } |
| 124 | |
| 125 | func (c *historicalImportCoordinator) loadQueueLocked() { |
| 126 | c.queueLoaded = true |
| 127 | saved, err := readHistoricalSidecar() |
| 128 | if err != nil { |
| 129 | return |
| 130 | } |
| 131 | c.current = "" |
| 132 | c.queue = append([]string{}, saved.Queue...) |
| 133 | if saved.Current != "" { |
| 134 | c.queue = append([]string{saved.Current}, c.queue...) |
| 135 | } |
| 136 | c.queueRevision, c.presentations = saved.QueueRevision, saved.Presentations |
| 137 | c.paused = len(c.queue) > 0 |
| 138 | } |
| 139 | |
| 140 | func (c *historicalImportCoordinator) saveQueueLocked() error { |
| 141 | var revision uint64 |
| 142 | err := updateHistoricalSidecar(func(saved *historicalImportQueueSidecar) error { |
| 143 | if saved.QueueRevision != c.queueRevision { |
| 144 | return errors.New("historical batch changed in another instance; refresh and retry") |
| 145 | } |
| 146 | saved.QueueRevision++ |
| 147 | revision = saved.QueueRevision |
| 148 | saved.Current, saved.Queue = c.current, append([]string{}, c.queue...) |
| 149 | return nil |
| 150 | }) |
| 151 | if err == nil { |
| 152 | c.queueRevision = revision |
| 153 | } |
| 154 | return err |
| 155 | } |
| 156 | |
| 157 | func (c *historicalImportCoordinator) claimQueueLocked() error { |
| 158 | if c.queueRelease != nil { |
| 159 | return nil |
| 160 | } |
| 161 | path := historicalImportQueuePath() |
| 162 | if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil { |
| 163 | return err |
| 164 | } |
| 165 | release, err := identitylock.TryAcquire(path + ".worker.lock") |
| 166 | if err != nil { |
| 167 | return errors.New("historical batch is active in another instance") |
| 168 | } |
| 169 | c.queueRelease = release |
| 170 | c.loadQueueLocked() |
| 171 | return nil |
| 172 | } |
| 173 | |
| 174 | func (c *historicalImportCoordinator) releaseQueueLocked() { |
| 175 | if c.queueRelease != nil { |
| 176 | c.queueRelease() |
| 177 | c.queueRelease = nil |
| 178 | } |
| 179 | } |
| 180 |