db.gno
19.29 Kb · 637 lines
1package radio
2
3import (
4 "strings"
5
6 "gno.land/p/nym-alexiscolin000/gnoradio/blocks/v0"
7 "gno.land/p/nym-alexiscolin000/gnoradio/store/v0"
8 "gno.land/r/nym-alexiscolin000/gnoradio/data"
9)
10
11// The radio's collections in the data realm (docs/ARCHITECTURE-v1.md,
12// section 3). P(n) is store.Pad, T(t) a 12-digit unix time. Numbers inside
13// the hot records are fixed-width base-64 digits (store.Num, store.Fixed): a decimal
14// conversion costs the VM several times more.
15const (
16 cStations = "radio/stations" // P(station) -> Rec(epoch, ring, live, rotation header, schedule, recent picks, weekly top[, ring rotation, ring index])
17 cRot = "radio/rot" // P(station)/b/P(block) -> block; P(station)/g/P(group) -> block sums (p/blocks)
18 cHomes = "radio/homes" // P((id-1)/homesChunk) -> Rec of homes entries (the genre stations of a track), one per synced track
19 cMeta = "radio/meta" // "flow" -> flow record
20 cCurators = "radio/curators" // address -> curator record (curators.gno)
21 cTops = "radio/tops" // "all" -> this week's top curators across stations (a station's is in its record)
22 cDropped = "radio/dropped" // P(station)/P(track) -> "1"
23 cNotes = "radio/notes" // P(station)/T(start) -> reporters, or the strike a hide gave (notes.gno)
24 cMuted = "radio/muted" // address -> unix time its dedications are refused until
25 cStrikes = "radio/strikes" // address -> unix time one of its dedications was last hidden
26 cSponsor = "radio/sponsor" // address/P(station) -> open sponsored pick (sponsor.gno)
27 cActivity = "radio/activity" // P(0..63) -> activity record (a ring, its head in the flow record)
28 cConfig = "radio/config" // "admin" (mirror of the admin role), "modbot" (hex key)
29)
30
31var collections = []string{cStations, cRot, cHomes, cMeta, cCurators, cTops, cDropped, cNotes,
32 cMuted, cStrikes, cSponsor, cActivity, cConfig}
33
34const (
35 homesChunk = 16 // homes entries per key
36 maxRecord = 16 << 10 // the data realm's longest value
37)
38
39// setup makes the radio's collections and its first records, once: a later
40// release finds them made (it is not the writer in its init).
41func setup(cur realm, holder address) {
42 for _, c := range collections {
43 data.Make(cross(cur), c)
44 }
45 if data.Has(cMeta, "flow") {
46 return
47 }
48 ops := store.Ops{}.Set(cMeta, "flow", strings.Repeat("0", flowLen)).Set(cConfig, "admin", holder.String())
49 data.Batch(cross(cur), ops)
50}
51
52// Fixed-width base-64 numbers are store.Num, store.Fixed and
53// store.AppendFixed. dec is store.Digits' decoding table, for the digits
54// the hot loops read in place: a call per digit costs about 1% of a pick.
55var dec = decTable()
56
57func decTable() (d [128]int64) {
58 for i := 0; i < len(store.Digits); i++ {
59 d[store.Digits[i]] = int64(i)
60 }
61 return d
62}
63
64// ---- tx: every key read once and written once ----
65
66// tx reads each key from the data realm once and writes every key it
67// changed once, in one call (data.Set or data.Batch); the stations it
68// loaded are decoded once and encoded back when changed. Every entrypoint
69// loads what it needs, changes it in its tx, then saves; readers use one
70// too. A tx is never stored.
71type tx struct {
72 es []entry
73 idx map[string]int // entry key -> position in es
74 sts [numStations]*station
75 flw string // the flow record, once read (flowRec)
76}
77
78type entry struct {
79 c, k string // collection and key
80 val string
81 ok bool // the key exists
82 dirty bool
83}
84
85func rd() *tx { return &tx{idx: map[string]int{}} }
86
87// at finds a key's entry, reading it from data the first time unless blind
88// (a key about to be overwritten).
89func (t *tx) at(c, k string, blind bool) *entry {
90 key := c + " " + k
91 if i, ok := t.idx[key]; ok {
92 return &t.es[i]
93 }
94 e := entry{c: c, k: k}
95 if !blind {
96 e.val, e.ok = data.Get(c, k)
97 }
98 t.idx[key] = len(t.es)
99 t.es = append(t.es, e)
100 return &t.es[len(t.es)-1]
101}
102
103func (t *tx) get(c, k string) (string, bool) {
104 e := t.at(c, k, false)
105 return e.val, e.ok
106}
107
108func (t *tx) val(c, k string) string { return t.at(c, k, false).val }
109
110func (t *tx) has(c, k string) bool { return t.at(c, k, false).ok }
111
112func (t *tx) set(c, k, v string) {
113 e := t.at(c, k, false)
114 e.val, e.ok, e.dirty = v, true, true
115}
116
117// put writes a key without reading it first.
118func (t *tx) put(c, k, v string) {
119 e := t.at(c, k, true)
120 e.val, e.ok, e.dirty = v, true, true
121}
122
123func (t *tx) del(c, k string) bool {
124 e := t.at(c, k, false)
125 if !e.ok {
126 return false
127 }
128 e.val, e.ok, e.dirty = "", false, true
129 return true
130}
131
132// save writes what t changed: one data.Set or data.Remove, or Batches of
133// 64 ops (a Sync or a RefreshArtist may change more keys than one Batch
134// takes; the transaction keeps them atomic).
135func save(cur realm, t *tx) {
136 for _, st := range t.sts {
137 if st != nil && st.dirty {
138 sched := st.raw
139 if st.decoded {
140 sched = encodeSchedule(st.sched)
141 }
142 f := []string{store.Fixed(st.epoch, 6), store.Fixed(int64(st.ring), 6), store.Fixed(int64(st.live), 4), st.h, sched, st.picks, st.top}
143 if st.ringed() {
144 f = append(f, st.rot, st.idx)
145 }
146 // A pick is refused once its station's record would leave less
147 // than recordHeadroom, and the writes a pick or a Sync makes on
148 // other stations wait while theirs does (full): this is a last
149 // resort.
150 rec := store.Rec(f...)
151 if len(rec) > maxRecord {
152 panic("radio: this station is full for now, try again later")
153 }
154 t.set(cStations, store.Pad(st.id), rec)
155 st.dirty = false
156 }
157 }
158 var ops store.Ops
159 for _, e := range t.es {
160 if !e.dirty {
161 continue
162 }
163 if e.ok {
164 ops = ops.Set(e.c, e.k, e.val)
165 } else {
166 ops = ops.Del(e.c, e.k)
167 }
168 }
169 for len(ops) > 4*64 {
170 data.Batch(cross(cur), ops[:4*64])
171 ops = ops[4*64:]
172 }
173 switch {
174 case len(ops) == 0:
175 case len(ops) > 4:
176 data.Batch(cross(cur), ops)
177 case ops[0] == store.OpSet:
178 data.Set(cross(cur), ops[1], ops[2], ops[3])
179 default:
180 data.Remove(cross(cur), ops[1], ops[2])
181 }
182}
183
184// ---- the flow record: radio/meta "flow" ----
185
186// The flow record is fixed-width: synced (4 digits), the number of dropped
187// slots (4), the activity ring's next slot (1), the week of the top across
188// stations and its last score when it holds topCurators (0 else) (4 and
189// 4: a pick that scores less leaves it as it is, unread), then "1" for each
190// genre station with a live track ("0" else, index = genre): Main relays
191// one of those (hourGenre). Every call reads it.
192const (
193 fSynced = 0
194 fDrops = 4
195 fHead = 8
196 fTopWeek = 9
197 fTopLast = 13
198 fLive = 17
199 flowLen = fLive + NumGenres + 1
200)
201
202// flowRec is the flow record, read once per tx: every call reads it. The data
203// realm is permanent: a record an earlier release wrote shorter is padded
204// with "0" (a later one's extra fields are kept as they are).
205func (t *tx) flowRec() string {
206 if t.flw == "" {
207 t.flw = t.val(cMeta, "flow")
208 if len(t.flw) < flowLen {
209 t.flw += strings.Repeat("0", flowLen-len(t.flw))
210 }
211 }
212 return t.flw
213}
214
215func (t *tx) flowAt(at, w int) int { return int(store.Num(t.flowRec(), at, w)) }
216
217func (t *tx) setFlowAt(at, w, v int) {
218 f := t.flowRec()
219 if n := store.Fixed(int64(v), w); n != f[at:at+w] {
220 t.flw = f[:at] + n + f[at+w:]
221 t.set(cMeta, "flow", t.flw)
222 }
223}
224
225func (t *tx) synced() int { return t.flowAt(fSynced, 4) }
226
227// hasLive reports whether a genre station has a live track.
228func (t *tx) hasLive(genre int) bool {
229 return genre >= 1 && genre <= NumGenres && t.flowRec()[fLive+genre] == '1'
230}
231
232// dropped reports whether the admin dropped a track from a station.
233func (t *tx) dropped(stationID, trackID int) bool {
234 return t.flowAt(fDrops, 4) > 0 && t.has(cDropped, dropKey(stationID, trackID))
235}
236
237func dropKey(stationID, trackID int) string { return store.Pad(stationID) + "/" + store.Pad(trackID) }
238
239// ---- stations ----
240
241// station is a station loaded in a tx: its header and schedule are one
242// record (cStations); its rotation is read and written a block at a time
243// (cRot), but for New and Choice: their rings (at most 500 slots) and the
244// index of the tracks in them are part of their record, which a pick
245// changes anyway.
246// The schedule is decoded on first use (slots): a call that only needs the
247// time paused under it (paused) or a picker (picker) reads it as it is.
248type station struct {
249 t *tx
250 id int
251 genre int // 0 = every genre
252 epoch int64
253 ring int // New and Choice: joins so far; the slot reused next is ring % keep
254 live int // rotation slots with a duration > 0
255 h string // the rotation's header (p/blocks)
256 raw string // the encoded schedule
257 picks string // the picks of the last replayGap (lastPicked)
258 top string // this week's top curators here (curators.gno)
259 rot string // New and Choice: the group then the blocks of their ring
260 idx string // and the ring's tracks: sorted entries, the track (4 digits) and its slot (2)
261 sched []Slot // sorted by Start, short (folded as it ages), once decoded
262 decoded bool
263 dirty bool
264 pz int64 // paused(pzAt), while pzOK: the schedule did not move since
265 pzAt int64
266 pzOK bool
267}
268
269// station loads a station once per tx.
270func (t *tx) station(id int) *station {
271 if id < 0 || id >= numStations {
272 panic("radio: unknown station")
273 }
274 if st := t.sts[id]; st != nil {
275 return st
276 }
277 st := &station{t: t, id: id, genre: id}
278 if id > NumGenres {
279 st.genre = 0
280 }
281 if rec := t.val(cStations, store.Pad(id)); rec != "" {
282 f := store.Fields(rec)
283 st.epoch, st.ring, st.live, st.h, st.raw, st.picks, st.top = store.Num(f[0], 0, 6), int(store.Num(f[1], 0, 6)), int(store.Num(f[2], 0, 4)), f[3], f[4], f[5], f[6]
284 if len(f) > 7 {
285 st.rot, st.idx = f[7], f[8]
286 }
287 }
288 t.sts[id] = st
289 return st
290}
291
292// size is the length of the record save writes for the station, its
293// schedule measured as encodeSchedule writes it, without writing it.
294func (st *station) size() int {
295 n := 1 + 3*7 + 6 + 6 + 4 + len(st.h) + len(st.picks) + len(st.top)
296 if st.ringed() {
297 n += 3*2 + len(st.rot) + len(st.idx)
298 }
299 if !st.decoded {
300 return n + len(st.raw)
301 }
302 var prevEnd int64
303 for i := range st.sched {
304 s := &st.sched[i]
305 switch gap := s.Start - prevEnd; {
306 case gap == 0:
307 n++
308 case gap < 0:
309 n += 7
310 default:
311 n += 2
312 for gap>>6 != 0 {
313 gap >>= 6
314 n++
315 }
316 }
317 n += 25 + len(s.By) + len(s.Note)
318 prevEnd = s.Start + s.Dur
319 }
320 return n
321}
322
323// full reports whether the record keeps less than recordHeadroom: a new pick
324// is refused, and a write a pick makes on the side (a Choice or New join)
325// waits. A side write adds far less than recordHeadroom, so
326// the record never reaches the data realm's limit (maxRecord).
327func (st *station) full() bool { return st.size() > maxRecord-recordHeadroom }
328
329// slots is the decoded schedule.
330func (st *station) slots() []Slot {
331 if !st.decoded {
332 st.sched, st.decoded = decodeSchedule(st.raw), true
333 }
334 return st.sched
335}
336
337// setSlots replaces the schedule.
338func (st *station) setSlots(s []Slot) {
339 st.sched, st.decoded, st.dirty, st.pzOK = s, true, true, false
340}
341
342// addLive counts a slot going live (d = 1) or silent (-1); the flow record
343// keeps which genre stations have one.
344func (st *station) addLive(d int) {
345 st.live += d
346 st.dirty = true
347 if st.id >= 1 && st.id <= NumGenres && (st.live > 0) != st.t.hasLive(st.id) {
348 on := 0
349 if st.live > 0 {
350 on = 1
351 }
352 st.t.setFlowAt(fLive+st.id, 1, on)
353 }
354}
355
356// ---- a station's rotation: p/blocks over its keys ----
357
358func (st *station) ringed() bool { return st.id == NewStation || st.id == ChoiceStation }
359
360// A ring's rotation (one group) is its group's sums, 4 digits per block,
361// then its blocks, blocks.Size slots each but the last.
362const blockLen = blocks.Size * 6
363
364// ringBlocks is where a ring's blocks start.
365func (st *station) ringBlocks() int { return 4 * ((st.count() + blocks.Size - 1) / blocks.Size) }
366
367// group and block read group g and block b of the rotation.
368func (st *station) group(g int) string {
369 if st.ringed() {
370 return st.rot[:st.ringBlocks()]
371 }
372 return st.t.val(cRot, store.Pad(st.id)+"/g/"+store.Pad(g))
373}
374
375func (st *station) block(b int) string {
376 if !st.ringed() {
377 return st.t.val(cRot, store.Pad(st.id)+"/b/"+store.Pad(b))
378 }
379 at := st.ringBlocks() + b*blockLen
380 if at >= len(st.rot) {
381 return ""
382 }
383 if at+blockLen < len(st.rot) {
384 return st.rot[at : at+blockLen]
385 }
386 return st.rot[at:]
387}
388
389// setRotVal writes group g and block b; st.h is still the rotation's
390// header before the change.
391func (st *station) setRotVal(g int, grp string, b int, blk string) {
392 if !st.ringed() {
393 st.t.set(cRot, store.Pad(st.id)+"/g/"+store.Pad(g), grp)
394 st.t.set(cRot, store.Pad(st.id)+"/b/"+store.Pad(b), blk)
395 return
396 }
397 bs := st.rot[st.ringBlocks():]
398 at, end := b*blockLen, b*blockLen+blockLen
399 if at > len(bs) {
400 at = len(bs)
401 }
402 if end > len(bs) {
403 end = len(bs)
404 }
405 st.rot, st.dirty = grp+bs[:at]+blk+bs[end:], true
406}
407
408func (st *station) total() int64 { return blocks.Total(st.h) }
409func (st *station) count() int { return blocks.Len(st.h) }
410
411func (st *station) rotSlot(i int) (int, int64) {
412 if i < 0 || i >= st.count() {
413 panic("radio: rotation index out of range")
414 }
415 _, b := blocks.Where(i)
416 return blocks.Slot(st.block(b), i)
417}
418
419// find returns the rotation slot playing at position pos of the loop.
420func (st *station) find(pos int64) (slot, id int, offset int64, ok bool) {
421 g, rest, ok := blocks.FindGroup(st.h, pos)
422 if !ok {
423 return 0, 0, 0, false
424 }
425 b, rest := blocks.FindBlock(st.group(g), g, rest)
426 slot, id, offset = blocks.FindSlot(st.block(b), b, rest)
427 return slot, id, offset, true
428}
429
430func (st *station) prefixBefore(i int) int64 {
431 g, b := blocks.Where(i)
432 return blocks.PrefixBefore(st.h, st.group(g), st.block(b), i)
433}
434
435// appendRot adds a slot at the end of the rotation and returns its index.
436func (st *station) appendRot(id int, dur int64) int {
437 g, b := blocks.Where(st.count())
438 h, grp, blk, i := blocks.Append(st.h, st.group(g), st.block(b), id, dur)
439 st.setRotVal(g, grp, b, blk)
440 st.h, st.dirty = h, true
441 return i
442}
443
444// setRot replaces slot i's id and duration.
445func (st *station) setRot(i, id int, dur int64) {
446 g, b := blocks.Where(i)
447 h, grp, blk := blocks.Set(st.h, st.group(g), st.block(b), i, id, dur)
448 st.setRotVal(g, grp, b, blk)
449 st.h, st.dirty = h, true
450}
451
452// ---- schedules ----
453
454// A schedule is its slots in order, each a record that starts with the gap
455// since the previous slot's end (one digit "0" when it follows it, else the
456// number of digits and the gap, "z" and the start if it overlaps), then its
457// body: "L" (listener), the duration (2 digits), the track (4), the artist
458// (4), the time booked (6), the refund (4), "1" when its note is hidden, and
459// its address and note, each after its length (1 and 2 digits). A body is reused as long as its slot keeps
460// its duration and note, so moving slots together costs no encoding.
461
462func decodeSchedule(s string) []Slot {
463 out := make([]Slot, 0, len(s)/8+1)
464 for c := (cursor{s: s}); c.next(); {
465 b := s[c.b:c.i]
466 x := Slot{Track: int(dec[b[3]]<<18 | dec[b[4]]<<12 | dec[b[5]]<<6 | dec[b[6]]), Start: c.start, Dur: c.end - c.start, body: b, bdur: c.end - c.start}
467 bl := int(dec[b[22]])
468 x.artist, x.At, x.Pay, x.NoteHidden = int(store.Num(b, 7, 4)), store.Num(b, 11, 6), store.Num(b, 17, 4), b[21] == '1'
469 x.By, x.Note = address(b[23:23+bl]), b[25+bl:]
470 out = append(out, x)
471 }
472 return out
473}
474
475func encodeSchedule(ss []Slot) string {
476 b := make([]byte, 0, 8*len(ss)+64)
477 var prevEnd int64
478 for i := range ss {
479 s := &ss[i]
480 switch {
481 case s.body == "":
482 s.body = s.encodeBody()
483 case s.bdur != s.Dur:
484 s.body = s.body[:1] + string(store.AppendFixed(make([]byte, 0, 2), s.Dur, 2)) + s.body[3:]
485 }
486 s.bdur = s.Dur
487 switch gap := s.Start - prevEnd; {
488 case gap == 0:
489 b = append(b, '0')
490 case gap < 0:
491 b = store.AppendFixed(append(b, 'z'), s.Start, 6)
492 default:
493 n := 1
494 for gap>>(6*n) != 0 {
495 n++
496 }
497 b = store.AppendFixed(append(b, store.Digits[n]), gap, n)
498 }
499 b = append(b, s.body...)
500 prevEnd = s.Start + s.Dur
501 }
502 return string(b)
503}
504
505func (s *Slot) encodeBody() string {
506 b := make([]byte, 0, 32+len(s.By)+len(s.Note))
507 b = store.AppendFixed(store.AppendFixed(store.AppendFixed(append(b, 'L'), s.Dur, 2), int64(s.Track), 4), int64(s.artist), 4)
508 b = store.AppendFixed(store.AppendFixed(b, s.At, 6), s.Pay, 4)
509 if s.NoteHidden {
510 b = append(b, '1')
511 } else {
512 b = append(b, '0')
513 }
514 b = append(store.AppendFixed(b, int64(len(s.By)), 1), s.By...)
515 return string(append(store.AppendFixed(b, int64(len(s.Note)), 2), s.Note...))
516}
517
518// cursor walks an encoded schedule a slot at a time without decoding it:
519// after next, the slot's body is s[b:i] and it plays from start to end.
520type cursor struct {
521 s string
522 i, b int
523 start, end int64
524}
525
526func (c *cursor) next() bool {
527 s, i := c.s, c.i
528 if i >= len(s) {
529 return false
530 }
531 start := c.end
532 switch ch := s[i]; ch {
533 case '0':
534 i++
535 case 'z':
536 start, i = store.Num(s, i+1, 6), i+7
537 default:
538 n := int(dec[s[i]])
539 start, i = c.end+store.Num(s, i+1, n), i+1+n
540 }
541 c.b, c.start, c.end = i, start, start+(dec[s[i+1]]<<6|dec[s[i+2]])
542 k := i + 23 + int(dec[s[i+22]])
543 c.i = k + 2 + int(dec[s[k]]<<6|dec[s[k+1]])
544 return true
545}
546
547// ---- where a track rotates: Main at id-1, New and Choice in their ring
548// index, genre stations in homes ----
549
550const idxW = 6 // a ring index entry
551
552// ringFind returns the position of a track in the ring index, or where it goes.
553func (st *station) ringFind(trackID int) (int, bool) {
554 lo, hi := 0, len(st.idx)/idxW
555 for lo < hi {
556 mid := (lo + hi) / 2
557 if int(store.Num(st.idx, idxW*mid, 4)) < trackID {
558 lo = mid + 1
559 } else {
560 hi = mid
561 }
562 }
563 return lo, lo < len(st.idx)/idxW && int(store.Num(st.idx, idxW*lo, 4)) == trackID
564}
565
566// ringSlot is a track's slot in the ring.
567func (st *station) ringSlot(trackID int) (int, bool) {
568 p, ok := st.ringFind(trackID)
569 if !ok {
570 return 0, false
571 }
572 return int(store.Num(st.idx, idxW*p+4, 2)), true
573}
574
575// ringIndex files a track at a slot of the ring, at position p of the
576// index (ringFind), replacing the entry there when found.
577func (st *station) ringIndex(p int, found bool, trackID, slot int) {
578 end := idxW * p
579 if found {
580 end += idxW
581 }
582 st.idx, st.dirty = st.idx[:idxW*p]+store.Fixed(int64(trackID), 4)+store.Fixed(int64(slot), 2)+st.idx[end:], true
583}
584
585// ringUnindex forgets a track that left the ring's slot and returns where
586// it was in the index (-1: it was not there).
587func (st *station) ringUnindex(trackID, slot int) int {
588 if p, ok := st.ringFind(trackID); ok && int(store.Num(st.idx, idxW*p+4, 2)) == slot {
589 st.idx, st.dirty = st.idx[:idxW*p]+st.idx[idxW*p+idxW:], true
590 return p
591 }
592 return -1
593}
594
595// A homes entry is 5 digits per genre station: the station (1) and the slot (4).
596
597func homesKey(trackID int) (string, int) {
598 return store.Pad((trackID - 1) / homesChunk), (trackID - 1) % homesChunk
599}
600
601// homes returns a synced track's entry.
602func (t *tx) homes(trackID int) string {
603 k, i := homesKey(trackID)
604 if ch := t.val(cHomes, k); ch != "" {
605 return store.Field(ch, i)
606 }
607 return ""
608}
609
610func (t *tx) setHomes(trackID int, e string) {
611 k, i := homesKey(trackID)
612 t.set(cHomes, k, store.With(t.val(cHomes, k), i, e))
613}
614
615// appendHomes files the entry of the next synced track.
616func (t *tx) appendHomes(trackID int, e string) {
617 k, i := homesKey(trackID)
618 if i == 0 {
619 t.put(cHomes, k, store.Rec(e))
620 return
621 }
622 t.set(cHomes, k, store.Append(t.val(cHomes, k), e))
623}
624
625func homeEntry(stationID, slot int) string {
626 return store.Fixed(int64(stationID), 1) + store.Fixed(int64(slot), 4)
627}
628
629// homeSlot finds a station's slot in an entry.
630func homeSlot(e string, stationID int) (int, bool) {
631 for i := 0; i+5 <= len(e); i += 5 {
632 if int(dec[e[i]]) == stationID {
633 return int(store.Num(e, i+1, 4)), true
634 }
635 }
636 return 0, false
637}
