diff --git a/machinery/cmd/mp4analyze/main.go b/machinery/cmd/mp4analyze/main.go new file mode 100644 index 0000000..e51689a --- /dev/null +++ b/machinery/cmd/mp4analyze/main.go @@ -0,0 +1,425 @@ +package main + +import ( + "fmt" + "os" + + "github.com/Eyevinn/mp4ff/avc" + mp4ff "github.com/Eyevinn/mp4ff/mp4" +) + +func main() { + if len(os.Args) < 2 { + fmt.Println("usage: mp4analyze ") + os.Exit(1) + } + f, err := os.Open(os.Args[1]) + if err != nil { + panic(err) + } + defer f.Close() + parsed, err := mp4ff.DecodeFile(f) + if err != nil { + panic(err) + } + + // Movie-level info + if parsed.Init != nil && parsed.Init.Moov != nil { + moov := parsed.Init.Moov + fmt.Printf("ftyp/moov present. timescale(mvhd)=%d duration(mvhd)=%d\n", + moov.Mvhd.Timescale, moov.Mvhd.Duration) + for _, trak := range moov.Traks { + ts := trak.Mdia.Mdhd.Timescale + fmt.Printf(" trak id=%d handler=%s mdhd.timescale=%d mdhd.duration=%d\n", + trak.Tkhd.TrackID, trak.Mdia.Hdlr.HandlerType, ts, trak.Mdia.Mdhd.Duration) + } + } else { + fmt.Println("no Init/Moov (pure fragmented stream?)") + } + + // sidx vs actual segment layout. MSE players use sidx to map presentation + // time -> byte ranges; if sidx references disagree with the real segment + // sizes/durations (e.g. after an early/short flush) the player fetches the + // wrong bytes and fails to decode — a failure that "heals" on seek. + fmt.Println("=== sidx references vs actual segments ===") + var sidxRefs []mp4ff.SidxRef + for _, c := range parsed.Children { + if s, ok := c.(*mp4ff.SidxBox); ok { + fmt.Printf(" sidx: timescale=%d earliestPresTime=%d firstOffset=%d refCount=%d anchor(after sidx)=%d\n", + s.Timescale, s.EarliestPresentationTime, s.FirstOffset, len(s.SidxRefs), s.AnchorPoint) + sidxRefs = s.SidxRefs + } + } + // Actual segment sizes (styp+moof+mdat) and fragment durations. + type segInfo struct { + size uint64 + dur uint64 + } + var actual []segInfo + for _, seg := range parsed.Segments { + var sz uint64 + if seg.Styp != nil { + sz += seg.Styp.Size() + } + if seg.Sidx != nil { + sz += seg.Sidx.Size() + } + var dur uint64 + for _, fr := range seg.Fragments { + sz += fr.Moof.Size() + if fr.Mdat != nil { + sz += fr.Mdat.Size() + } + for _, traf := range fr.Moof.Trafs { + if traf.Tfhd.TrackID != 1 { + continue + } + for _, trun := range traf.Truns { + for _, s := range trun.Samples { + dur += uint64(s.Dur) + } + } + } + } + actual = append(actual, segInfo{size: sz, dur: dur}) + } + for i := range actual { + refStr := "(no sidx ref)" + if i < len(sidxRefs) { + r := sidxRefs[i] + mark := "" + if uint64(r.ReferencedSize) != actual[i].size { + mark += fmt.Sprintf(" SIZE MISMATCH actual=%d", actual[i].size) + } + if uint64(r.SubSegmentDuration) != actual[i].dur { + mark += fmt.Sprintf(" DUR MISMATCH actual=%d", actual[i].dur) + } + refStr = fmt.Sprintf("sidx.size=%d sidx.dur=%d type=%d sap=%d/%d%s", + r.ReferencedSize, r.SubSegmentDuration, r.ReferenceType, r.StartsWithSAP, r.SAPType, mark) + } + fmt.Printf(" seg%02d actual.size=%d actual.dur=%d | %s\n", i, actual[i].size, actual[i].dur, refStr) + } + + fmt.Println("=== fragments ===") + fragIdx := 0 + var allKeyGlobal []uint64 // global keyframe decode times (track timescale units) + var prevTfdtEnd = map[uint32]uint64{} + for si, seg := range parsed.Segments { + for _, fr := range seg.Fragments { + for _, traf := range fr.Moof.Trafs { + tid := traf.Tfhd.TrackID + tfdt := traf.Tfdt.BaseMediaDecodeTime() + offset := uint64(0) + var keys []uint64 // keyframe offset-from-tfdt + var durs []uint64 + zeroDur := 0 + nSamples := 0 + for _, trun := range traf.Truns { + for _, s := range trun.Samples { + nSamples++ + if (s.Flags>>24)&0x03 == 0x02 { // sample_depends_on==2 => IDR/sync + keys = append(keys, offset) + if tid == 1 { + allKeyGlobal = append(allKeyGlobal, tfdt+offset) + } + } + if s.Dur == 0 { + zeroDur++ + } + durs = append(durs, uint64(s.Dur)) + offset += uint64(s.Dur) + } + } + cont := "" + if pe, ok := prevTfdtEnd[tid]; ok { + if tfdt != pe { + cont = fmt.Sprintf(" <-- tfdt GAP/JUMP prev_end=%d delta=%d", pe, int64(tfdt)-int64(pe)) + } + } + prevTfdtEnd[tid] = tfdt + offset + if tid == 1 { + // in-fragment keyframe gaps + var gaps []int64 + for i := 1; i < len(keys); i++ { + gaps = append(gaps, int64(keys[i])-int64(keys[i-1])) + } + fmt.Printf("seg%d frag%d trk%d tfdt=%d dur=%d nSamp=%d zeroDur=%d keys=%v inFragKeyGaps=%v%s\n", + si, fragIdx, tid, tfdt, offset, nSamples, zeroDur, keys, gaps, cont) + } + } + fragIdx++ + } + } + + fmt.Println("=== global video keyframe decode times & gaps ===") + for i, k := range allKeyGlobal { + gap := int64(0) + if i > 0 { + gap = int64(k) - int64(allKeyGlobal[i-1]) + } + flag := "" + if i > 1 { + prevGap := int64(allKeyGlobal[i-1]) - int64(allKeyGlobal[i-2]) + if gap > 0 && prevGap > 0 && gap*2 < prevGap { + flag = fmt.Sprintf(" <== SEAM? gap=%d < prevGap/2=%d", gap, prevGap/2) + } + } + fmt.Printf(" kf#%02d dt=%d gap=%d%s\n", i, k, gap, flag) + } + + // Full sample timeline: DTS, CTS (=DTS+cto), composition offset, NAL types, + // to detect PTS non-monotonicity / gaps / param-set changes at the seam. + fmt.Println("=== per-sample timeline (full) — checking PTS monotonicity & nal types ===") + var trex *mp4ff.TrexBox + if parsed.Init != nil && parsed.Init.Moov != nil && parsed.Init.Moov.Mvex != nil { + for _, t := range parsed.Init.Moov.Mvex.Trexs { + if t.TrackID == 1 { + trex = t + } + } + } + var lastCTS int64 = -1 + var lastDTS int64 = -1 + sampIdx := 0 + fragIdx = 0 + for _, seg := range parsed.Segments { + for _, fr := range seg.Fragments { + fs, err := fr.GetFullSamples(trex) + if err != nil { + fmt.Printf(" frag%d GetFullSamples err: %v\n", fragIdx, err) + fragIdx++ + continue + } + for _, s := range fs { + dts := int64(s.DecodeTime) + cts := dts + int64(s.CompositionTimeOffset) + nals := nalTypes(s.Data) + anomaly := "" + if lastCTS >= 0 && cts < lastCTS { + anomaly += fmt.Sprintf(" <== CTS BACKWARDS (prev=%d)", lastCTS) + } + if lastDTS >= 0 && dts < lastDTS { + anomaly += fmt.Sprintf(" <== DTS BACKWARDS (prev=%d)", lastDTS) + } + isSync := s.Flags&0x02000000 == 0 && (s.Flags>>24)&0x03 == 0x02 + // Only print near the seam region and any anomalies, to keep output small. + near := dts >= 7800 && dts <= 8700 + if near || anomaly != "" { + fmt.Printf(" s%04d frag%d dts=%d cts=%d cto=%d dur=%d size=%d sync=%v nal=%v%s\n", + sampIdx, fragIdx, dts, cts, s.CompositionTimeOffset, s.Dur, len(s.Data), isSync, nals, anomaly) + } + lastCTS = cts + lastDTS = dts + sampIdx++ + } + fragIdx++ + } + } + + // Compare parameter sets: avcC (in moov) vs inline SPS/PPS at every IDR. + // A looping source that restarts may re-emit SPS/PPS that differ from the + // ones the player configured its decoder with from avcC — a classic cause + // of a freeze that "heals" when you seek past the seam. + fmt.Println("=== parameter set comparison (avcC vs inline IDR) ===") + var avccSPS, avccPPS [][]byte + if parsed.Init != nil && parsed.Init.Moov != nil { + for _, trak := range parsed.Init.Moov.Traks { + if trak.Mdia == nil || trak.Mdia.Minf == nil || trak.Mdia.Minf.Stbl == nil { + continue + } + stsd := trak.Mdia.Minf.Stbl.Stsd + if stsd == nil || stsd.AvcX == nil || stsd.AvcX.AvcC == nil { + continue + } + avccSPS = stsd.AvcX.AvcC.SPSnalus + avccPPS = stsd.AvcX.AvcC.PPSnalus + } + } + for i, s := range avccSPS { + fmt.Printf(" avcC SPS[%d] = %x\n", i, s) + } + for i, p := range avccPPS { + fmt.Printf(" avcC PPS[%d] = %x\n", i, p) + } + fragIdx = 0 + sampIdx = 0 + var baseSPS, basePPS []byte + if len(avccSPS) > 0 { + baseSPS = avccSPS[0] + } + if len(avccPPS) > 0 { + basePPS = avccPPS[0] + } + for _, seg := range parsed.Segments { + for _, fr := range seg.Fragments { + fs, err := fr.GetFullSamples(trex) + if err != nil { + fragIdx++ + continue + } + for _, s := range fs { + spsList := nalsByType(s.Data, 7) + ppsList := nalsByType(s.Data, 8) + if len(spsList) > 0 || len(ppsList) > 0 { + dts := int64(s.DecodeTime) + note := "" + if len(spsList) > 0 { + if baseSPS == nil { + baseSPS = spsList[0] + } else if !bytesEqual(baseSPS, spsList[0]) { + note += " <== SPS CHANGED vs base/avcC" + } + } + if len(ppsList) > 0 { + if basePPS == nil { + basePPS = ppsList[0] + } else if !bytesEqual(basePPS, ppsList[0]) { + note += " <== PPS CHANGED vs base/avcC" + } + } + var spsHex, ppsHex string + if len(spsList) > 0 { + spsHex = fmt.Sprintf("%x", spsList[0]) + } + if len(ppsList) > 0 { + ppsHex = fmt.Sprintf("%x", ppsList[0]) + } + fmt.Printf(" IDR s%04d frag%d dts=%d SPS=%s PPS=%s%s\n", + sampIdx, fragIdx, dts, spsHex, ppsHex, note) + } + sampIdx++ + } + fragIdx++ + } + } + + sliceHeaders(parsed, trex) +} + +func sliceHeaders(parsed *mp4ff.File, trex *mp4ff.TrexBox) { + // Build SPS/PPS maps from avcC. + spsMap := map[uint32]*avc.SPS{} + ppsMap := map[uint32]*avc.PPS{} + if parsed.Init != nil && parsed.Init.Moov != nil { + for _, trak := range parsed.Init.Moov.Traks { + if trak.Mdia == nil || trak.Mdia.Minf == nil || trak.Mdia.Minf.Stbl == nil { + continue + } + stsd := trak.Mdia.Minf.Stbl.Stsd + if stsd == nil || stsd.AvcX == nil || stsd.AvcX.AvcC == nil { + continue + } + for _, s := range stsd.AvcX.AvcC.SPSnalus { + if sps, err := avc.ParseSPSNALUnit(s, true); err == nil { + spsMap[uint32(sps.ParameterID)] = sps + } + } + for _, p := range stsd.AvcX.AvcC.PPSnalus { + if pps, err := avc.ParsePPSNALUnit(p, spsMap); err == nil { + ppsMap[pps.PicParameterSetID] = pps + } + } + } + } + + fmt.Println("=== slice headers near seam (frame_num / poc / idr_pic_id) ===") + fragIdx := 0 + sampIdx := 0 + for _, seg := range parsed.Segments { + for _, fr := range seg.Fragments { + fs, err := fr.GetFullSamples(trex) + if err != nil { + fragIdx++ + continue + } + for _, s := range fs { + dts := int64(s.DecodeTime) + if dts < 6800 || dts > 9400 { + sampIdx++ + continue + } + for _, nal := range splitAVCC(s.Data) { + t := nal[0] & 0x1f + if t == 1 || t == 5 { // non-IDR or IDR slice + sh, err := avc.ParseSliceHeader(nal, spsMap, ppsMap) + if err != nil { + fmt.Printf(" s%04d frag%d dts=%d nalType=%d sliceHeader ERR: %v\n", sampIdx, fragIdx, dts, t, err) + break + } + fmt.Printf(" s%04d frag%d dts=%d nalType=%d sliceType=%v frameNum=%d idrPicId=%d pocLsb=%d\n", + sampIdx, fragIdx, dts, t, sh.SliceType, sh.FrameNum, sh.IDRPicID, sh.PicOrderCntLsb) + break + } + } + sampIdx++ + } + fragIdx++ + } + } +} + +// splitAVCC splits a length-prefixed (4-byte) AVCC buffer into NAL units. +func splitAVCC(b []byte) [][]byte { + var out [][]byte + i := 0 + for i+4 <= len(b) { + n := int(uint32(b[i])<<24 | uint32(b[i+1])<<16 | uint32(b[i+2])<<8 | uint32(b[i+3])) + i += 4 + if n <= 0 || i+n > len(b) { + break + } + out = append(out, b[i:i+n]) + i += n + } + return out +} + +func bytesEqual(a, b []byte) bool { + if len(a) != len(b) { + return false + } + for i := range a { + if a[i] != b[i] { + return false + } + } + return true +} + +// nalTypes returns the list of H.264 NAL unit types present in an AVCC +// (length-prefixed) sample buffer. +func nalTypes(b []byte) []int { + var out []int + i := 0 + for i+4 <= len(b) { + n := int(uint32(b[i])<<24 | uint32(b[i+1])<<16 | uint32(b[i+2])<<8 | uint32(b[i+3])) + i += 4 + if n <= 0 || i+n > len(b) { + break + } + out = append(out, int(b[i]&0x1f)) + i += n + } + return out +} + +// nalsByType returns the raw NAL payloads (without length prefix) of the given +// type from an AVCC (length-prefixed) sample buffer. +func nalsByType(b []byte, want int) [][]byte { + var out [][]byte + i := 0 + for i+4 <= len(b) { + n := int(uint32(b[i])<<24 | uint32(b[i+1])<<16 | uint32(b[i+2])<<8 | uint32(b[i+3])) + i += 4 + if n <= 0 || i+n > len(b) { + break + } + if int(b[i]&0x1f) == want { + nal := make([]byte, n) + copy(nal, b[i:i+n]) + out = append(out, nal) + } + i += n + } + return out +} diff --git a/machinery/src/cloud/kerberos_vault.go b/machinery/src/cloud/kerberos_vault.go index a64d3b1..76070fe 100644 --- a/machinery/src/cloud/kerberos_vault.go +++ b/machinery/src/cloud/kerberos_vault.go @@ -41,17 +41,6 @@ func UploadKerberosVault(configuration *models.Configuration, fileName string) ( // KerberosCloud, this means storage is disabled and proxy enabled. log.Log.Info("UploadKerberosVault: Uploading to Kerberos Vault (" + config.KStorage.URI + ")") log.Log.Info("UploadKerberosVault: Upload started for " + fileName) - fullname := "data/recordings/" + fileName - - file, err := os.OpenFile(fullname, os.O_RDWR, 0755) - if file != nil { - defer file.Close() - } - if err != nil { - err := "UploadKerberosVault: Upload Failed, file doesn't exists anymore" - log.Log.Info(err) - return false, false, errors.New(err) - } publicKey := config.KStorage.CloudKey if config.HubKey != "" { @@ -60,62 +49,30 @@ func UploadKerberosVault(configuration *models.Configuration, fileName string) ( // We need to check if we are in a retry timeout. if kstorageRetryTimeout <= time.Now().Unix() { + uploaded, responded, body, err := sendToVault(*config.KStorage, publicKey, config.Key, fileName, "UploadKerberosVault", "primary") + if uploaded { + kstorageRetryCount = 0 + log.Log.Info("UploadKerberosVault: Upload Finished, " + body) + return true, true, nil + } - req, err := http.NewRequest("POST", config.KStorage.URI+"/storage", file) if err != nil { - errorMessage := "UploadKerberosVault: error reading request, " + config.KStorage.URI + "/storage: " + err.Error() - log.Log.Error(errorMessage) - return false, true, errors.New(errorMessage) - } - req.Header.Set("Content-Type", "video/mp4") - req.Header.Set("X-Kerberos-Storage-CloudKey", publicKey) - req.Header.Set("X-Kerberos-Storage-AccessKey", config.KStorage.AccessKey) - req.Header.Set("X-Kerberos-Storage-SecretAccessKey", config.KStorage.SecretAccessKey) - req.Header.Set("X-Kerberos-Storage-Provider", config.KStorage.Provider) - req.Header.Set("X-Kerberos-Storage-FileName", fileName) - req.Header.Set("X-Kerberos-Storage-Device", config.Key) - req.Header.Set("X-Kerberos-Storage-Capture", "IPCamera") - req.Header.Set("X-Kerberos-Storage-Directory", config.KStorage.Directory) - - var client *http.Client - if os.Getenv("AGENT_TLS_INSECURE") == "true" { - tr := &http.Transport{ - TLSClientConfig: &tls.Config{InsecureSkipVerify: true}, - } - client = &http.Client{Transport: tr} - } else { - client = &http.Client{} - } - - resp, err := client.Do(req) - if resp != nil { - defer resp.Body.Close() - } - - if err == nil { - if resp != nil { - body, err := io.ReadAll(resp.Body) - if err == nil { - if resp.StatusCode == 200 { - kstorageRetryCount = 0 - log.Log.Info("UploadKerberosVault: Upload Finished, " + resp.Status + ", " + string(body)) - return true, true, nil - } else { - // We increase the retry count, and set the timeout. - // If we have reached the retry policy, we set the timeout. - // This means we will not retry for the next 5 minutes. - if kstorageRetryCount < config.KStorage.MaxRetries { - kstorageRetryCount = (kstorageRetryCount + 1) - } - if kstorageRetryCount == config.KStorage.MaxRetries { - kstorageRetryTimeout = time.Now().Add(time.Duration(config.KStorage.Timeout) * time.Second).Unix() - } - log.Log.Info("UploadKerberosVault: Upload Failed, " + resp.Status + ", " + string(body)) - } - } - } - } else { log.Log.Info("UploadKerberosVault: Upload Failed, " + err.Error()) + } else { + log.Log.Info("UploadKerberosVault: Upload Failed, " + body) + } + + // We only advance the retry policy when the vault gave a definitive + // response (mirroring the original behaviour where transient network + // errors did not consume retries). When the retry count reaches the + // configured maximum we back off for the configured timeout. + if responded { + if kstorageRetryCount < config.KStorage.MaxRetries { + kstorageRetryCount = (kstorageRetryCount + 1) + } + if kstorageRetryCount == config.KStorage.MaxRetries { + kstorageRetryTimeout = time.Now().Add(time.Duration(config.KStorage.Timeout) * time.Second).Unix() + } } } @@ -134,61 +91,111 @@ func UploadKerberosVault(configuration *models.Configuration, fileName string) ( log.Log.Info("UploadKerberosVault (Secondary): Uploading to Secondary Kerberos Vault (" + config.KStorageSecondary.URI + ")") - file, err = os.OpenFile(fullname, os.O_RDWR, 0755) - if file != nil { - defer file.Close() - } - if err != nil { - err := "UploadKerberosVault (Secondary): Upload Failed, file doesn't exists anymore" - log.Log.Info(err) - return false, false, errors.New(err) + uploaded, _, body, err := sendToVault(*config.KStorageSecondary, publicKey, config.Key, fileName, "UploadKerberosVault (Secondary)", "secondary") + if uploaded { + log.Log.Info("UploadKerberosVault (Secondary): Upload Finished to secondary, " + body) + return true, true, nil } - req, err := http.NewRequest("POST", config.KStorageSecondary.URI+"/storage", file) if err != nil { - errorMessage := "UploadKerberosVault (Secondary): error reading request, " + config.KStorageSecondary.URI + "/storage: " + err.Error() - log.Log.Error(errorMessage) - return false, true, errors.New(errorMessage) - } - req.Header.Set("Content-Type", "video/mp4") - req.Header.Set("X-Kerberos-Storage-CloudKey", publicKey) - req.Header.Set("X-Kerberos-Storage-AccessKey", config.KStorageSecondary.AccessKey) - req.Header.Set("X-Kerberos-Storage-SecretAccessKey", config.KStorageSecondary.SecretAccessKey) - req.Header.Set("X-Kerberos-Storage-Provider", config.KStorageSecondary.Provider) - req.Header.Set("X-Kerberos-Storage-FileName", fileName) - req.Header.Set("X-Kerberos-Storage-Device", config.Key) - req.Header.Set("X-Kerberos-Storage-Capture", "IPCamera") - req.Header.Set("X-Kerberos-Storage-Directory", config.KStorageSecondary.Directory) - - var client *http.Client - if os.Getenv("AGENT_TLS_INSECURE") == "true" { - tr := &http.Transport{ - TLSClientConfig: &tls.Config{InsecureSkipVerify: true}, - } - client = &http.Client{Transport: tr} + log.Log.Info("UploadKerberosVault (Secondary): Upload Failed to secondary, " + err.Error()) } else { - client = &http.Client{} - } - - resp, err := client.Do(req) - if resp != nil { - defer resp.Body.Close() - } - - if err == nil { - if resp != nil { - body, err := io.ReadAll(resp.Body) - if err == nil { - if resp.StatusCode == 200 { - log.Log.Info("UploadKerberosVault (Secondary): Upload Finished to secondary, " + resp.Status + ", " + string(body)) - return true, true, nil - } else { - log.Log.Info("UploadKerberosVault (Secondary): Upload Failed to secondary, " + resp.Status + ", " + string(body)) - } - } - } + log.Log.Info("UploadKerberosVault (Secondary): Upload Failed to secondary, " + body) } } return false, true, nil } + +// sendToVault uploads a single recording to one Kerberos Vault. When resumable +// uploads are enabled (the default) it attempts the tus protocol first and, if +// the vault does not expose a tus endpoint (older deployments), transparently +// falls back to the legacy single-shot POST. +// +// It returns whether the upload succeeded, whether the vault gave a definitive +// HTTP response (so the caller can advance its retry policy), a short message +// for logging, and a transport error if any. +func sendToVault(vault models.KStorage, publicKey, deviceKey, fileName, label, slot string) (bool, bool, string, error) { + if resumableUploadsEnabled() { + uploaded, responded, supported, body, err := uploadVaultResumable(vault, publicKey, deviceKey, fileName, label, slot) + if supported { + return uploaded, responded, body, err + } + log.Log.Info(label + ": resumable (tus) endpoint not available, falling back to legacy upload") + } + return uploadVaultLegacy(vault, publicKey, deviceKey, fileName, label) +} + +// uploadVaultLegacy performs the original single-request upload: the whole file +// is sent as the body of a POST to {URI}/storage. Kept for backwards +// compatibility with vault deployments that do not support resumable uploads. +func uploadVaultLegacy(vault models.KStorage, publicKey, deviceKey, fileName, label string) (bool, bool, string, error) { + fullname := "data/recordings/" + fileName + + file, err := os.OpenFile(fullname, os.O_RDWR, 0755) + if file != nil { + defer file.Close() + } + if err != nil { + msg := label + ": Upload Failed, file doesn't exists anymore" + log.Log.Info(msg) + return false, false, "", errors.New(msg) + } + + req, err := http.NewRequest("POST", vault.URI+"/storage", file) + if err != nil { + errorMessage := label + ": error reading request, " + vault.URI + "/storage: " + err.Error() + log.Log.Error(errorMessage) + return false, false, "", errors.New(errorMessage) + } + req.Header.Set("Content-Type", "video/mp4") + setVaultHeaders(req.Header, vault, publicKey, deviceKey, fileName) + + client := newVaultHTTPClient(0) + resp, err := client.Do(req) + if resp != nil { + defer resp.Body.Close() + } + if err != nil { + return false, false, "", err + } + + body, rerr := io.ReadAll(resp.Body) + if rerr != nil { + return false, false, "", rerr + } + + if resp.StatusCode == 200 { + return true, true, resp.Status + ", " + string(body), nil + } + return false, true, resp.Status + ", " + string(body), nil +} + +// setVaultHeaders sets the standard Kerberos Vault headers used by the legacy +// single-POST upload. +func setVaultHeaders(h http.Header, vault models.KStorage, publicKey, deviceKey, fileName string) { + h.Set("X-Kerberos-Storage-CloudKey", publicKey) + h.Set("X-Kerberos-Storage-AccessKey", vault.AccessKey) + h.Set("X-Kerberos-Storage-SecretAccessKey", vault.SecretAccessKey) + h.Set("X-Kerberos-Storage-Provider", vault.Provider) + h.Set("X-Kerberos-Storage-FileName", fileName) + h.Set("X-Kerberos-Storage-Device", deviceKey) + h.Set("X-Kerberos-Storage-Capture", "IPCamera") + h.Set("X-Kerberos-Storage-Directory", vault.Directory) +} + +// newVaultHTTPClient builds an HTTP client honouring the AGENT_TLS_INSECURE +// escape hatch. A timeout of 0 disables the client-level timeout, which is +// required for streaming large upload bodies. +func newVaultHTTPClient(timeout time.Duration) *http.Client { + client := &http.Client{} + if os.Getenv("AGENT_TLS_INSECURE") == "true" { + client.Transport = &http.Transport{ + TLSClientConfig: &tls.Config{InsecureSkipVerify: true}, + } + } + if timeout > 0 { + client.Timeout = timeout + } + return client +} diff --git a/machinery/src/cloud/tus_client.go b/machinery/src/cloud/tus_client.go new file mode 100644 index 0000000..d88dcfb --- /dev/null +++ b/machinery/src/cloud/tus_client.go @@ -0,0 +1,396 @@ +package cloud + +import ( + "encoding/base64" + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "net/url" + "os" + "path/filepath" + "sort" + "strconv" + "strings" + "time" + + "github.com/kerberos-io/agent/machinery/src/log" + "github.com/kerberos-io/agent/machinery/src/models" +) + +// tusResumableVersion is the tus protocol version implemented by this client. +const tusResumableVersion = "1.0.0" + +// tusUploadPath is appended to the configured Kerberos Vault URI to reach the +// resumable upload endpoint. It mirrors how the legacy uploader appends +// "/storage". +const tusUploadPath = "/storage/tus/" + +// tusResumeState is persisted in a sidecar file next to the agent data so an +// interrupted upload can be resumed across retries and even agent restarts. +type tusResumeState struct { + UploadURL string `json:"upload_url"` + VaultURI string `json:"vault_uri"` + Size int64 `json:"size"` +} + +// resumableUploadsEnabled reports whether the resumable (tus) upload path should +// be attempted. It is enabled by default and can be disabled (falling back to +// the legacy single POST) by setting AGENT_DISABLE_RESUMABLE_UPLOAD=true. +func resumableUploadsEnabled() bool { + return os.Getenv("AGENT_DISABLE_RESUMABLE_UPLOAD") != "true" +} + +// uploadVaultResumable uploads a recording to a Kerberos Vault using the tus +// resumable upload protocol. +// +// Return values: +// - uploaded: the recording was fully received and persisted by the vault. +// - responded: the vault returned a definitive HTTP response (used by the +// caller to advance its retry/secondary-failover policy). +// - supported: the vault exposes a tus endpoint. When false, the caller should +// fall back to the legacy single-POST upload (older vault deployments). +// - body: a short message for logging. +func uploadVaultResumable(vault models.KStorage, publicKey, deviceKey, fileName, label, slot string) (uploaded bool, responded bool, supported bool, body string, err error) { + fullname := "data/recordings/" + fileName + + file, ferr := os.Open(fullname) + if file != nil { + defer file.Close() + } + if ferr != nil { + msg := label + ": resumable upload failed, file doesn't exist anymore" + log.Log.Info(msg) + // The file is gone, so the legacy path cannot help either. Report it as + // "supported" to avoid a pointless fallback attempt. + return false, false, true, "", errors.New(msg) + } + + info, serr := file.Stat() + if serr != nil { + return false, false, true, "", serr + } + size := info.Size() + + baseURL := strings.TrimRight(vault.URI, "/") + tusUploadPath + client := newVaultHTTPClient(0) + + metadata := encodeTusMetadata(map[string]string{ + "filename": fileName, + "device": deviceKey, + "directory": vault.Directory, + "provider": vault.Provider, + "capture": "IPCamera", + "cloudkey": publicKey, + }) + + sidecar := tusSidecarPath(fileName, slot) + uploadURL := loadTusResumeState(sidecar, baseURL) + + const maxAttempts = 4 + restartedAfterComplete := false + + for attempt := 0; attempt < maxAttempts; attempt++ { + // (1) Ensure we have an active upload URL, creating one if needed. + if uploadURL == "" { + created, status, cerr := tusCreate(client, baseURL, size, metadata, vault, publicKey, deviceKey, fileName) + if cerr != nil { + if status == http.StatusNotFound || status == http.StatusMethodNotAllowed || status == http.StatusNotImplemented { + // The vault does not implement tus; let the caller fall back. + return false, false, false, "", cerr + } + log.Log.Info(label + ": resumable create failed, " + cerr.Error()) + tusBackoff(attempt) + continue + } + uploadURL = created + saveTusResumeState(sidecar, tusResumeState{UploadURL: uploadURL, VaultURI: baseURL, Size: size}) + } + + // (2) Query the current server-side offset. + offset, status, herr := tusHead(client, uploadURL, vault, publicKey, deviceKey) + if herr != nil { + if status == http.StatusNotFound || status == http.StatusGone { + // The upload expired/was removed server-side; start over. + removeTusResumeState(sidecar) + uploadURL = "" + continue + } + log.Log.Info(label + ": resumable head failed, " + herr.Error()) + tusBackoff(attempt) + continue + } + + // (3) All bytes are present but the upload was not finalized (e.g. the + // completion hook failed). A completed tus upload cannot be re-finalized + // with another PATCH, so delete it and re-upload to force a clean finalize. + if offset >= size { + if restartedAfterComplete { + return false, true, true, "resumable finalize did not complete", errors.New(label + ": resumable finalize did not complete") + } + tusTerminate(client, uploadURL, vault, publicKey, deviceKey) + removeTusResumeState(sidecar) + uploadURL = "" + restartedAfterComplete = true + continue + } + + // (4) Stream the remaining bytes in a single PATCH directly from disk. + if _, sErr := file.Seek(offset, io.SeekStart); sErr != nil { + return false, false, true, "", sErr + } + newOffset, status, respBody, perr := tusPatch(client, uploadURL, offset, size, file, vault, publicKey, deviceKey) + if perr != nil { + if status >= 400 { + // Definitive rejection (e.g. provider push failed during finalize). + // Re-evaluate via HEAD on the next iteration to decide retry/restart. + log.Log.Info(label + ": resumable patch rejected, " + perr.Error()) + } else { + log.Log.Info(label + ": resumable patch failed, " + perr.Error()) + } + tusBackoff(attempt) + continue + } + if newOffset >= size { + removeTusResumeState(sidecar) + return true, true, true, respBody, nil + } + + // Partial progress: persist and continue with the next chunk. + saveTusResumeState(sidecar, tusResumeState{UploadURL: uploadURL, VaultURI: baseURL, Size: size}) + } + + return false, true, true, "resumable upload did not complete after retries", errors.New(label + ": resumable upload did not complete after retries") +} + +// tusCreate performs the tus "creation" request (POST). On success it returns +// the resolved upload URL the agent should use for subsequent HEAD/PATCH calls. +func tusCreate(client *http.Client, baseURL string, size int64, metadata string, vault models.KStorage, publicKey, deviceKey, fileName string) (string, int, error) { + req, err := http.NewRequest("POST", baseURL, nil) + if err != nil { + return "", 0, err + } + req.Header.Set("Tus-Resumable", tusResumableVersion) + req.Header.Set("Upload-Length", strconv.FormatInt(size, 10)) + if metadata != "" { + req.Header.Set("Upload-Metadata", metadata) + } + setVaultTusHeaders(req.Header, vault, publicKey, deviceKey, fileName) + + resp, err := client.Do(req) + if resp != nil { + defer resp.Body.Close() + } + if err != nil { + return "", 0, err + } + io.Copy(io.Discard, resp.Body) + + if resp.StatusCode != http.StatusCreated { + return "", resp.StatusCode, fmt.Errorf("unexpected status creating upload: %s", resp.Status) + } + location := resp.Header.Get("Location") + if location == "" { + return "", resp.StatusCode, errors.New("missing Location header in create response") + } + return resolveTusLocation(baseURL, location), resp.StatusCode, nil +} + +// tusHead performs the tus "offset" request (HEAD) and returns the current +// server-side upload offset. +func tusHead(client *http.Client, uploadURL string, vault models.KStorage, publicKey, deviceKey string) (int64, int, error) { + req, err := http.NewRequest("HEAD", uploadURL, nil) + if err != nil { + return 0, 0, err + } + req.Header.Set("Tus-Resumable", tusResumableVersion) + setVaultTusHeaders(req.Header, vault, publicKey, deviceKey, "") + + resp, err := client.Do(req) + if resp != nil { + defer resp.Body.Close() + } + if err != nil { + return 0, 0, err + } + io.Copy(io.Discard, resp.Body) + + if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusNoContent { + return 0, resp.StatusCode, fmt.Errorf("unexpected status on HEAD: %s", resp.Status) + } + offsetStr := resp.Header.Get("Upload-Offset") + offset, perr := strconv.ParseInt(offsetStr, 10, 64) + if perr != nil { + return 0, resp.StatusCode, fmt.Errorf("invalid Upload-Offset header: %q", offsetStr) + } + return offset, resp.StatusCode, nil +} + +// tusPatch streams the remaining bytes of the file (from offset to size) to the +// upload URL using a single PATCH request. The body is read straight from the +// *os.File, so the recording is never fully buffered in memory. +func tusPatch(client *http.Client, uploadURL string, offset, size int64, file io.Reader, vault models.KStorage, publicKey, deviceKey string) (int64, int, string, error) { + remaining := size - offset + req, err := http.NewRequest("PATCH", uploadURL, io.LimitReader(file, remaining)) + if err != nil { + return offset, 0, "", err + } + req.ContentLength = remaining + req.Header.Set("Tus-Resumable", tusResumableVersion) + req.Header.Set("Content-Type", "application/offset+octet-stream") + req.Header.Set("Upload-Offset", strconv.FormatInt(offset, 10)) + setVaultTusHeaders(req.Header, vault, publicKey, deviceKey, "") + + resp, err := client.Do(req) + if resp != nil { + defer resp.Body.Close() + } + if err != nil { + return offset, 0, "", err + } + bodyBytes, _ := io.ReadAll(resp.Body) + respBody := string(bodyBytes) + + if resp.StatusCode != http.StatusNoContent { + return offset, resp.StatusCode, respBody, fmt.Errorf("unexpected status on PATCH: %s, %s", resp.Status, respBody) + } + newOffsetStr := resp.Header.Get("Upload-Offset") + newOffset, perr := strconv.ParseInt(newOffsetStr, 10, 64) + if perr != nil { + // A 204 without a parseable offset means the upload finished. + return size, resp.StatusCode, respBody, nil + } + return newOffset, resp.StatusCode, respBody, nil +} + +// tusTerminate best-effort deletes an upload server-side (DELETE). +func tusTerminate(client *http.Client, uploadURL string, vault models.KStorage, publicKey, deviceKey string) { + req, err := http.NewRequest("DELETE", uploadURL, nil) + if err != nil { + return + } + req.Header.Set("Tus-Resumable", tusResumableVersion) + setVaultTusHeaders(req.Header, vault, publicKey, deviceKey, "") + + resp, derr := client.Do(req) + if resp != nil { + io.Copy(io.Discard, resp.Body) + resp.Body.Close() + } + _ = derr +} + +// setVaultTusHeaders sets the Kerberos Vault authentication and routing headers +// on every tus request. Credentials are sent on each request (and never stored +// server-side in the upload metadata). When fileName is empty it is omitted, as +// it is only useful on the creation request (routing also travels in the tus +// Upload-Metadata). +func setVaultTusHeaders(h http.Header, vault models.KStorage, publicKey, deviceKey, fileName string) { + h.Set("X-Kerberos-Storage-CloudKey", publicKey) + h.Set("X-Kerberos-Storage-AccessKey", vault.AccessKey) + h.Set("X-Kerberos-Storage-SecretAccessKey", vault.SecretAccessKey) + h.Set("X-Kerberos-Storage-Provider", vault.Provider) + h.Set("X-Kerberos-Storage-Device", deviceKey) + h.Set("X-Kerberos-Storage-Directory", vault.Directory) + h.Set("X-Kerberos-Storage-Capture", "IPCamera") + if fileName != "" { + h.Set("X-Kerberos-Storage-FileName", fileName) + } +} + +// encodeTusMetadata serializes a map into the tus Upload-Metadata header format: +// a comma separated list of "key base64(value)" pairs. Keys are sorted for a +// deterministic header value. Empty values are skipped. +func encodeTusMetadata(pairs map[string]string) string { + parts := make([]string, 0, len(pairs)) + for k, v := range pairs { + if v == "" { + continue + } + parts = append(parts, k+" "+base64.StdEncoding.EncodeToString([]byte(v))) + } + sort.Strings(parts) + return strings.Join(parts, ",") +} + +// resolveTusLocation turns the Location header returned by the create request +// into an absolute URL. To keep talking to the agent's configured vault host +// (and avoid issues when the vault sits behind a proxy that rewrites the host), +// it keeps the configured base URL and only appends the server-assigned upload +// id taken from the Location. +func resolveTusLocation(baseURL, location string) string { + if ref, err := url.Parse(location); err == nil { + trimmed := strings.Trim(ref.Path, "/") + if trimmed != "" { + segments := strings.Split(trimmed, "/") + id := segments[len(segments)-1] + if id != "" { + return strings.TrimRight(baseURL, "/") + "/" + id + } + } + } + // Fallback: resolve the reference against the base URL as-is. + if base, err := url.Parse(baseURL); err == nil { + if ref, err := url.Parse(location); err == nil { + return base.ResolveReference(ref).String() + } + } + return location +} + +// tusSidecarDir is the directory where resume state files are kept. It is +// intentionally separate from data/cloud (which is scanned for recordings to +// upload) so the sidecar files are never mistaken for recordings. +func tusSidecarDir() string { + return "data/tus" +} + +func tusSidecarPath(fileName, slot string) string { + safe := strings.ReplaceAll(fileName, "/", "_") + safe = strings.ReplaceAll(safe, string(os.PathSeparator), "_") + return filepath.Join(tusSidecarDir(), safe+"."+slot+".json") +} + +// loadTusResumeState returns a previously stored upload URL for the given +// sidecar, but only if it was created against the same vault base URL. Any +// mismatch or read/parse error yields an empty string (start fresh). +func loadTusResumeState(path, baseURL string) string { + b, err := os.ReadFile(path) + if err != nil { + return "" + } + var state tusResumeState + if err := json.Unmarshal(b, &state); err != nil { + return "" + } + if state.UploadURL == "" || state.VaultURI != baseURL { + return "" + } + return state.UploadURL +} + +func saveTusResumeState(path string, state tusResumeState) { + if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil { + return + } + b, err := json.Marshal(state) + if err != nil { + return + } + _ = os.WriteFile(path, b, 0o644) +} + +func removeTusResumeState(path string) { + _ = os.Remove(path) +} + +// tusBackoff sleeps for an exponentially increasing duration (capped) between +// resume attempts to avoid hammering a temporarily unavailable vault. +func tusBackoff(attempt int) { + delay := time.Duration(500*(1< 3*time.Second { + delay = 3 * time.Second + } + time.Sleep(delay) +} diff --git a/machinery/src/cloud/tus_client_test.go b/machinery/src/cloud/tus_client_test.go new file mode 100644 index 0000000..3736ab4 --- /dev/null +++ b/machinery/src/cloud/tus_client_test.go @@ -0,0 +1,347 @@ +package cloud + +import ( + "bytes" + "fmt" + "io" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "strconv" + "strings" + "sync" + "testing" + + "github.com/kerberos-io/agent/machinery/src/models" +) + +// fakeUpload tracks the state of a single resumable upload on the fake server. +type fakeUpload struct { + size int64 + offset int64 +} + +// fakeTus is a tiny in-memory implementation of the tus 1.0.0 server protocol, +// sufficient to exercise the agent's resumable client. +type fakeTus struct { + mu sync.Mutex + uploads map[string]*fakeUpload + counter int + creates int + lastPatchBytes int64 + + // unsupported makes the creation endpoint return 404, simulating an older + // vault without a tus endpoint. + unsupported bool + // failFinalize causes the next N completing PATCH requests to return 502 + // after storing the bytes, simulating a failed completion hook. + failFinalize int +} + +func newFakeTus() *fakeTus { + return &fakeTus{uploads: map[string]*fakeUpload{}} +} + +func (s *fakeTus) seed(size, offset int64) string { + s.mu.Lock() + defer s.mu.Unlock() + s.counter++ + id := fmt.Sprintf("seed-%d", s.counter) + s.uploads[id] = &fakeUpload{size: size, offset: offset} + return id +} + +func (s *fakeTus) totalBytes() int64 { + s.mu.Lock() + defer s.mu.Unlock() + var total int64 + for _, u := range s.uploads { + total += u.offset + } + return total +} + +func (s *fakeTus) lastPatch() int64 { + s.mu.Lock() + defer s.mu.Unlock() + return s.lastPatchBytes +} + +func (s *fakeTus) createCount() int { + s.mu.Lock() + defer s.mu.Unlock() + return s.creates +} + +func (s *fakeTus) ServeHTTP(w http.ResponseWriter, r *http.Request) { + id := strings.TrimPrefix(r.URL.Path, tusUploadPath) + w.Header().Set("Tus-Resumable", tusResumableVersion) + + switch r.Method { + case http.MethodPost: + if s.unsupported { + w.WriteHeader(http.StatusNotFound) + return + } + length, _ := strconv.ParseInt(r.Header.Get("Upload-Length"), 10, 64) + s.mu.Lock() + s.counter++ + s.creates++ + newID := fmt.Sprintf("up-%d", s.counter) + s.uploads[newID] = &fakeUpload{size: length} + s.mu.Unlock() + w.Header().Set("Location", tusUploadPath+newID) + w.WriteHeader(http.StatusCreated) + + case http.MethodHead: + s.mu.Lock() + u, ok := s.uploads[id] + s.mu.Unlock() + if !ok { + w.WriteHeader(http.StatusNotFound) + return + } + w.Header().Set("Upload-Offset", strconv.FormatInt(u.offset, 10)) + w.Header().Set("Upload-Length", strconv.FormatInt(u.size, 10)) + w.WriteHeader(http.StatusOK) + + case http.MethodPatch: + s.mu.Lock() + u, ok := s.uploads[id] + s.mu.Unlock() + if !ok { + w.WriteHeader(http.StatusNotFound) + return + } + n, _ := io.Copy(io.Discard, r.Body) + s.mu.Lock() + u.offset += n + s.lastPatchBytes = n + complete := u.offset >= u.size + failNow := complete && s.failFinalize > 0 + if failNow { + s.failFinalize-- + } + offset := u.offset + s.mu.Unlock() + + w.Header().Set("Upload-Offset", strconv.FormatInt(offset, 10)) + if failNow { + // Bytes are stored but the (simulated) completion hook failed. + w.WriteHeader(http.StatusBadGateway) + return + } + w.WriteHeader(http.StatusNoContent) + + case http.MethodDelete: + s.mu.Lock() + delete(s.uploads, id) + s.mu.Unlock() + w.WriteHeader(http.StatusNoContent) + + default: + w.WriteHeader(http.StatusMethodNotAllowed) + } +} + +// withRecording switches into a fresh temp working directory containing a +// recording at data/recordings/. The working directory is restored on +// cleanup. Tests using this helper must not run in parallel. +func withRecording(t *testing.T, fileName string, payload []byte) { + t.Helper() + dir := t.TempDir() + old, err := os.Getwd() + if err != nil { + t.Fatalf("getwd: %v", err) + } + if err := os.Chdir(dir); err != nil { + t.Fatalf("chdir: %v", err) + } + t.Cleanup(func() { _ = os.Chdir(old) }) + + if err := os.MkdirAll("data/recordings", 0o755); err != nil { + t.Fatalf("mkdir recordings: %v", err) + } + if err := os.WriteFile(filepath.Join("data/recordings", fileName), payload, 0o644); err != nil { + t.Fatalf("write recording: %v", err) + } +} + +func testVault(uri string) models.KStorage { + return models.KStorage{ + URI: uri, + AccessKey: "ak", + SecretAccessKey: "sk", + Provider: "gcp", + Directory: "dir", + } +} + +func TestUploadVaultResumable_HappyPath(t *testing.T) { + srv := newFakeTus() + ts := httptest.NewServer(srv) + defer ts.Close() + + fileName := "1564859471_6-474162_oprit_577-283-727-375_1153_27.mp4" + payload := bytes.Repeat([]byte("x"), 4096) + withRecording(t, fileName, payload) + + uploaded, responded, supported, _, err := uploadVaultResumable(testVault(ts.URL), "pk", "dev", fileName, "test", "primary") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if !uploaded || !responded || !supported { + t.Fatalf("uploaded/responded/supported = %v/%v/%v, want all true", uploaded, responded, supported) + } + if got := srv.totalBytes(); got != int64(len(payload)) { + t.Fatalf("server received %d bytes, want %d", got, len(payload)) + } + if _, err := os.Stat(tusSidecarPath(fileName, "primary")); !os.IsNotExist(err) { + t.Fatalf("expected sidecar to be removed after success, stat err = %v", err) + } +} + +func TestUploadVaultResumable_Unsupported(t *testing.T) { + srv := newFakeTus() + srv.unsupported = true + ts := httptest.NewServer(srv) + defer ts.Close() + + fileName := "f.mp4" + withRecording(t, fileName, []byte("hello")) + + uploaded, _, supported, _, _ := uploadVaultResumable(testVault(ts.URL), "pk", "dev", fileName, "test", "primary") + if uploaded { + t.Fatal("expected uploaded=false against a vault without a tus endpoint") + } + if supported { + t.Fatal("expected supported=false so the caller falls back to the legacy upload") + } +} + +func TestUploadVaultResumable_FinalizeRetry(t *testing.T) { + srv := newFakeTus() + srv.failFinalize = 1 + ts := httptest.NewServer(srv) + defer ts.Close() + + fileName := "1564859471_6-474162_oprit_577-283-727-375_1153_27.mp4" + payload := bytes.Repeat([]byte("y"), 2048) + withRecording(t, fileName, payload) + + uploaded, _, supported, _, err := uploadVaultResumable(testVault(ts.URL), "pk", "dev", fileName, "test", "primary") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if !uploaded || !supported { + t.Fatalf("expected success after a failed finalize + restart, got uploaded=%v supported=%v", uploaded, supported) + } + if got := srv.createCount(); got < 2 { + t.Fatalf("expected at least 2 create requests (restart after failed finalize), got %d", got) + } +} + +func TestUploadVaultResumable_ResumeFromSidecar(t *testing.T) { + srv := newFakeTus() + ts := httptest.NewServer(srv) + defer ts.Close() + + fileName := "1564859471_6-474162_oprit_577-283-727-375_1153_27.mp4" + total := 8192 + half := 4096 + payload := bytes.Repeat([]byte("z"), total) + withRecording(t, fileName, payload) + + // Simulate a previous run that uploaded half the file before being interrupted. + id := srv.seed(int64(total), int64(half)) + baseURL := strings.TrimRight(ts.URL, "/") + tusUploadPath + saveTusResumeState(tusSidecarPath(fileName, "primary"), tusResumeState{ + UploadURL: strings.TrimRight(baseURL, "/") + "/" + id, + VaultURI: baseURL, + Size: int64(total), + }) + + uploaded, _, supported, _, err := uploadVaultResumable(testVault(ts.URL), "pk", "dev", fileName, "test", "primary") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if !uploaded || !supported { + t.Fatalf("expected resume success, got uploaded=%v supported=%v", uploaded, supported) + } + if got := srv.lastPatch(); got != int64(total-half) { + t.Fatalf("resume should only send the remaining %d bytes, sent %d", total-half, got) + } + if srv.createCount() != 0 { + t.Fatalf("resume should not create a new upload, got %d creates", srv.createCount()) + } +} + +func TestEncodeTusMetadata(t *testing.T) { + got := encodeTusMetadata(map[string]string{ + "b": "2", + "a": "1", + "empty": "", + }) + // keys sorted, empty values skipped, values base64-encoded. + want := "a MQ==,b Mg==" + if got != want { + t.Fatalf("encodeTusMetadata = %q, want %q", got, want) + } +} + +func TestResolveTusLocation(t *testing.T) { + cases := []struct { + name string + base string + location string + want string + }{ + { + name: "absolute path location", + base: "http://host/storage/tus/", + location: "/storage/tus/abc", + want: "http://host/storage/tus/abc", + }, + { + name: "absolute url keeps configured host", + base: "http://host/storage/tus/", + location: "http://internal:8080/storage/tus/xyz", + want: "http://host/storage/tus/xyz", + }, + { + name: "relative id", + base: "http://host/api/storage/tus/", + location: "abc", + want: "http://host/api/storage/tus/abc", + }, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + if got := resolveTusLocation(tc.base, tc.location); got != tc.want { + t.Fatalf("resolveTusLocation(%q, %q) = %q, want %q", tc.base, tc.location, got, tc.want) + } + }) + } +} + +func TestTusResumeStateRoundTrip(t *testing.T) { + dir := t.TempDir() + old, _ := os.Getwd() + if err := os.Chdir(dir); err != nil { + t.Fatalf("chdir: %v", err) + } + defer os.Chdir(old) + + path := tusSidecarPath("file.mp4", "primary") + state := tusResumeState{UploadURL: "http://host/storage/tus/abc", VaultURI: "http://host/storage/tus/", Size: 123} + saveTusResumeState(path, state) + + if got := loadTusResumeState(path, state.VaultURI); got != state.UploadURL { + t.Fatalf("loadTusResumeState = %q, want %q", got, state.UploadURL) + } + // A mismatched vault URI must not be reused. + if got := loadTusResumeState(path, "http://other/storage/tus/"); got != "" { + t.Fatalf("loadTusResumeState with mismatched vault = %q, want empty", got) + } +}