cmd/streamdata: peek at subsequent assets for parts

Some streams disconnect half-way. This collects all parts.

Signed-off-by: Yonah <contrib@gensokyo.uk>

cmd/streamdata/format.go

@@ -18,4 +18,7 @@ func printVOD(v *streamdata.VOD) {
 	} else {
 		fmt.Printf("YouTube  : %s | %s\n", v.Title, date)
 	}
+	if v.Mirror != "" {
+		fmt.Println("Mirror   :", v.Mirror)
+	}
 }

cmd/streamdata/main.go

@@ -2,6 +2,7 @@ package main
 
 import (
 	"bufio"
+	"cmp"
 	"context"
 	"errors"
 	"fmt"
@@ -11,6 +12,8 @@ import (
 	"os"
 	"os/signal"
 	"path"
+	"slices"
+	"strconv"
 	"strings"
 	"syscall"
 	"time"
@@ -221,13 +224,108 @@ func main() {
 				return errors.New("list requires a channel selected")
 			}
 
-			for ident := range channel.All(&err) {
+			for ident := range channel.All(&err, true) {
 				fmt.Println(ident)
 			}
 			return
 		},
 	)
 
+	c.NewCommand(
+		"mirror",
+		"Populate mirror address of the next un-mirrored asset",
+		func([]string) (err error) {
+			if channel == nil {
+				return errors.New("mirror requires a channel selected")
+			}
+
+			idents := slices.Collect(channel.All(&err, false))
+			if err != nil {
+				return
+			}
+			slices.SortFunc(idents, func(a, b *streamdata.Ident) int {
+				return cmp.Compare(a.Serial, b.Serial)
+			})
+
+			unique := errors.New("entry already mirrored")
+			for i, ident := range idents {
+				if err = channel.Edit(ident, func(v *streamdata.VOD) (err error) {
+					if v.Mirror != "" {
+						return unique
+					}
+
+					var parts []*streamdata.VOD
+					for _, next := range idents[i+1:] {
+						var part *streamdata.VOD
+						if part, err = channel.Load(next); err != nil {
+							return
+						}
+						if part.Title != v.Title {
+							break
+						}
+						parts = append(parts, part)
+					}
+
+					pathname := path.Join(
+						flagBase,
+						"upload"+streamdata.ChannelVODSuffix,
+					)
+					if err = os.Symlink(path.Join(
+						"vod",
+						ident.String()+streamdata.ChannelVODSuffix,
+					), pathname); err != nil {
+						return
+					}
+					log.Println("pending video at", pathname)
+					defer func() {
+						removeErr := os.Remove(pathname)
+						if err == nil {
+							err = removeErr
+						}
+					}()
+
+					defer toErr(&err)
+					isattyStd()
+					br := bufio.NewReader(os.Stdin)
+
+					if len(parts) > 0 {
+						fmt.Println("assets with identical title:")
+						for j := range parts {
+							fmt.Println(idents[i+1+j])
+						}
+						if !require(promptBool(br, "Proceed? ", true)) {
+							for j, part := range parts {
+								fmt.Println(idents[i+1+j])
+								printVOD(part)
+							}
+							return errors.New(
+								strconv.Itoa(len(parts)) + " trailing parts",
+							)
+						}
+					}
+
+				retry:
+					fmt.Println(ident)
+					printVOD(v)
+					v.Mirror = require(promptTrimNoEmpty(br, "Mirror   : "))
+					if !require(promptBool(br, "Proceed? ", true)) {
+						v.Mirror = ""
+						goto retry
+					}
+					return nil
+				}); err != nil {
+					if err == unique {
+						err = nil
+						continue
+					}
+					return
+				}
+				break
+			}
+			return
+		},
+	)
+
 	c.MustParse(os.Args[1:], func(err error) {
 		if channel != nil {
 			if closeErr := channel.Close(); closeErr != nil {

streamdata.go

@@ -130,6 +130,8 @@ type VOD struct {
 	Date time.Time `json:"date"`
 	// Free-form category string.
 	Category string `json:"category,omitempty"`
+	// A mirror site the asset is uploaded to, usually YouTube.
+	Mirror string `json:"mirror,omitempty"`
 }
 
 // ChannelMismatchError describes a mismatching [Ident.Channel] passed to [Channel.Add].
@@ -143,6 +145,14 @@ func (c *ChannelMismatchError) Error() string {
 		strconv.FormatUint(c.Want, 10)
 }
 
+// RenameFrom is a pathname returned by a function passed to [Channel.Add] to
+// rename from this pathname instead of the managed transaction file.
+type RenameFrom string
+
+func (pathname RenameFrom) Error() string {
+	return "requesting rename from " + strconv.Quote(string(pathname))
+}
+
 // Add adds a [VOD] and its corresponding asset to the on-disk representation.
 func (c *Channel) Add(ident *Ident, f func(v *VOD, w io.Writer) error) error {
 	if ident == nil || f == nil {
@@ -167,8 +177,20 @@ func (c *Channel) Add(ident *Ident, f func(v *VOD, w io.Writer) error) error {
 		err = closeErr
 	}
 	if err != nil {
-		_ = c.root.Remove(channelPathPending)
+		var rf RenameFrom
-		return err
+		if !errors.As(err, &rf) {
+			_ = c.root.Remove(channelPathPending)
+			return err
+		}
+		if err = c.root.Remove(channelPathPending); err != nil {
+			return err
+		}
+		if err = os.Rename(string(rf), path.Join(
+			c.root.Name(),
+			channelPathPending,
+		)); err != nil {
+			return err
+		}
 	}
 
 	if err = c.root.Chmod(channelPathPending, 0444); err != nil {
@@ -202,6 +224,46 @@ func (c *Channel) Add(ident *Ident, f func(v *VOD, w io.Writer) error) error {
 	return nil
 }
 
+// Edit loads a [VOD] by its [Ident], and writes the modified [VOD] back to the
+// on-disk representation.
+func (c *Channel) Edit(ident *Ident, f func(v *VOD) error) error {
+	v, err := c.Load(ident)
+	if err != nil {
+		return err
+	}
+
+	err = f(v)
+	if err != nil {
+		return err
+	}
+
+	pathname := path.Join(
+		channelPathVOD,
+		"."+ident.String(),
+	)
+
+	var w *os.File
+	if w, err = c.root.OpenFile(
+		pathname,
+		os.O_WRONLY|os.O_CREATE|os.O_EXCL,
+		0444,
+	); err != nil {
+		return err
+	} else if err = json.NewEncoder(w).Encode(v); err != nil {
+		_ = w.Close()
+		_ = c.root.Remove(pathname)
+		return err
+	} else if err = w.Close(); err != nil {
+		_ = c.root.Remove(pathname)
+		return err
+	} else {
+		return c.root.Rename(pathname, path.Join(
+			channelPathVOD,
+			ident.String(),
+		))
+	}
+}
+
 // Path returns a pathname by [Ident].
 func (c *Channel) Path(ident *Ident) string {
 	return path.Join(
@@ -214,7 +276,10 @@ func (c *Channel) Path(ident *Ident) string {
 // All returns an iterator over all known [Ident] in the on-disk representation.
 // Iteration stops when encountering the first non-nil error, and its value is
 // saved to the value pointed to by errP.
-func (c *Channel) All(errP *error) iter.Seq[*Ident] {
+//
+// If reuse is true, the same value is updated every iteration and the same
+// address is yileded as a result.
+func (c *Channel) All(errP *error, reuse bool) iter.Seq[*Ident] {
 	return func(yield func(*Ident) bool) {
 		dents, err := c.root.FS().(fs.ReadDirFS).ReadDir(channelPathVOD)
 		if err != nil {
@@ -236,7 +301,11 @@ func (c *Channel) All(errP *error) iter.Seq[*Ident] {
 				return
 			}
 
-			if !yield(&ident) {
+			p := &ident
+			if !reuse {
+				p = new(ident)
+			}
+			if !yield(p) {
 				return
 			}
 		}

streamdata_test.go

@@ -194,7 +194,7 @@ func TestChannelBadMetadata(t *testing.T) {
 	})
 }
 
-func TestChannelAdd(t *testing.T) {
+func TestChannel(t *testing.T) {
 	t.Parallel()
 
 	d := t.TempDir()
@@ -303,7 +303,7 @@ func TestChannelAdd(t *testing.T) {
 	}
 
 	var iterErr error
-	idents := slices.Collect(c.All(&iterErr))
+	idents := slices.Collect(c.All(&iterErr, false))
 	if iterErr != nil {
 		t.Fatalf("All: error = %#v", iterErr)
 	}
@@ -334,6 +334,50 @@ func TestChannelAdd(t *testing.T) {
 		t.Errorf("Perm: %#o", fi.Mode().Perm())
 	}
 
+	rf := streamdata.RenameFrom(path.Join(d, "alternate"))
+	if err := os.WriteFile(string(rf), []byte{0}, 0); err != nil {
+		t.Fatal(err)
+	}
+	if err := c.Add(&streamdata.Ident{Channel: 0xcafe}, func(*streamdata.VOD, io.Writer) error {
+		return rf
+	}); err != nil {
+		t.Fatalf("Add: error = %v", err)
+	}
+	if _, err := os.Stat(string(rf)); !errors.Is(err, os.ErrNotExist) {
+		t.Fatalf("Stat: error = %v", err)
+	}
+	if data, err := os.ReadFile(path.Join(
+		d,
+		"vod",
+		(&streamdata.Ident{Channel: 0xcafe}).String()+".mp4",
+	)); err != nil {
+		t.Fatal(err)
+	} else if string(data) != "\x00" {
+		t.Errorf("(rf) Add: %#v", data)
+	}
+
+	wantVODEdit := streamdata.VOD{
+		Title:    "edit\x00",
+		Date:     time.Unix(0xdeadbeef, 0).UTC(),
+		Category: "\t\x00",
+		Mirror:   "https://satori.hifuu.internal/edit.mp4\x00",
+	}
+	if err := c.Edit(&streamdata.Ident{Channel: 0xcafe}, func(v *streamdata.VOD) error {
+		if *v != (streamdata.VOD{}) {
+			t.Errorf("Edit: v = %#v", v)
+		}
+
+		*v = wantVODEdit
+		return nil
+	}); err != nil {
+		t.Fatalf("Edit: error = %v", err)
+	}
+	if gotVODEdit, err := c.Load(&streamdata.Ident{Channel: 0xcafe}); err != nil {
+		t.Fatalf("(edit) Load: error = %v", err)
+	} else if *gotVODEdit != wantVODEdit {
+		t.Errorf("Edit: %#v, want %#v", *gotVODEdit, wantVODEdit)
+	}
+
 	if err := os.Chmod(path.Join(d, "vod", wantIdent), 0); err != nil {
 		t.Fatal(err)
 	}