Owner-flagged after the dev-docs audit: BroadcastProgressUpdate, BroadcastAnnotationUpdate, and BroadcastConflictNotification fanned out through the untargeted Broadcast, delivering one user's book progress, conflict notices, and — worst — annotation payloads (which can carry note/highlight text) to EVERY connected client on the instance. A family-lens violation of exactly the kind the RLS move eliminated for database reads: user B's client received events for user A's books, including books in libraries hidden from B. Fix is the RLS philosophy applied to the fan-out layer: enforcement at one choke point. The untargeted Broadcast method and its channel/loop machinery are deleted — BroadcastToUser is now the only door, and the compiler forces every future emitter to name its audience instead of relying on review. The three helpers take the acting user's ID; the four sync call sites (progress.go x2, annotations.go x2) already had the acting user in scope. Scan events were already per-triggering- identity and collections already per-user — unchanged. Logging is untouched: the only log line in the diff is the deleted method's own, so docker logs remains the complete administrator audit trail. TestBroadcastIsPerUser pins the invariant (alice's two devices receive her three events; bob's client receives nothing — any message there is a cross-user leak). Both websocket docs updated in the same commit: the wire doc's fan-out section now states per-user delivery and the chokepoint, and the gated endpoint doc's intro matches. Full make test green.
779 lines
27 KiB
Go
779 lines
27 KiB
Go
package sync
|
|
|
|
import (
|
|
"bookhoard/internal/database"
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"log"
|
|
"math"
|
|
"time"
|
|
|
|
"github.com/google/uuid"
|
|
"github.com/jackc/pgx/v5"
|
|
"github.com/jackc/pgx/v5/pgtype"
|
|
)
|
|
|
|
// ProgressData represents universal progress data with multiple location references
|
|
type ProgressData struct {
|
|
Percentage *float64 `json:"percentage,omitempty"`
|
|
Epubcfi *string `json:"epubcfi,omitempty"`
|
|
Character *int64 `json:"character,omitempty"`
|
|
Chapter *int `json:"chapter,omitempty"`
|
|
ChapterProgress *float64 `json:"chapter_progress,omitempty"`
|
|
ViewportY *float64 `json:"viewport_y,omitempty"`
|
|
Page *int `json:"page,omitempty"`
|
|
TotalPages *int `json:"total_pages,omitempty"`
|
|
PageY *int `json:"page_y,omitempty"`
|
|
Zoom *float64 `json:"zoom,omitempty"`
|
|
ScrollX *float64 `json:"scroll_x,omitempty"`
|
|
ScrollY *float64 `json:"scroll_y,omitempty"`
|
|
Panel *int `json:"panel,omitempty"`
|
|
ReadingMode *string `json:"reading_mode,omitempty"`
|
|
TotalCharacters *int64 `json:"total_characters,omitempty"`
|
|
// Audiobook locator: seconds into the current chapter. Pairs with
|
|
// Chapter; the anti-reset guard still owns pushes that would drag
|
|
// the percentage backwards.
|
|
AudioOffsetSeconds *int `json:"audio_offset_seconds,omitempty"`
|
|
}
|
|
|
|
// DeviceProgress represents progress from a specific device
|
|
type DeviceProgress struct {
|
|
Source string `json:"source"`
|
|
Data ProgressData `json:"data"`
|
|
Timestamp string `json:"timestamp,omitempty"`
|
|
}
|
|
|
|
// ConvertProgress converts progress between different format groups
|
|
func ConvertProgress(sourceFormat, targetFormat FormatGroup, sourceData map[string]interface{}) (map[string]interface{}, error) {
|
|
percentage := extractPercentage(sourceFormat, sourceData)
|
|
|
|
result := make(map[string]interface{})
|
|
|
|
switch targetFormat {
|
|
case FormatGroupReflowable:
|
|
result["percentage"] = percentage
|
|
if epubcfi, ok := sourceData["epubcfi"].(string); ok {
|
|
result["epubcfi"] = epubcfi
|
|
} else {
|
|
// Generate approximate CFI from percentage
|
|
result["epubcfi"] = fmt.Sprintf("epubcfi(/6/4/2:%d)", int(percentage*100))
|
|
}
|
|
if totalChars, ok := sourceData["total_characters"].(int64); ok {
|
|
result["character"] = int64(float64(totalChars) * percentage)
|
|
}
|
|
|
|
case FormatGroupFixedLayout:
|
|
totalPages := 200.0
|
|
if tp, ok := sourceData["total_pages"].(int); ok {
|
|
totalPages = float64(tp)
|
|
}
|
|
result["page"] = int(math.Round(percentage * totalPages))
|
|
result["total_pages"] = int(totalPages)
|
|
result["percentage"] = percentage
|
|
if pageY, ok := sourceData["page_y"].(int); ok {
|
|
result["page_y"] = pageY
|
|
}
|
|
|
|
case FormatGroupComicArchive:
|
|
totalPages := 32.0
|
|
if tp, ok := sourceData["total_pages"].(int); ok {
|
|
totalPages = float64(tp)
|
|
}
|
|
result["page"] = int(math.Round(percentage * totalPages))
|
|
result["total_pages"] = int(totalPages)
|
|
result["percentage"] = percentage
|
|
if panel, ok := sourceData["panel"].(int); ok {
|
|
result["panel"] = panel
|
|
}
|
|
|
|
case FormatGroupAudiobook:
|
|
// Audio progress is percentage + global seconds; there is no page
|
|
// grid to re-derive, so the percentage carries across as-is and
|
|
// the second-based locator follows when present.
|
|
result["percentage"] = percentage
|
|
if offset, ok := sourceData["audio_offset_seconds"].(int); ok {
|
|
result["audio_offset_seconds"] = offset
|
|
}
|
|
|
|
default:
|
|
return nil, fmt.Errorf("unsupported target format: %s", targetFormat)
|
|
}
|
|
|
|
return result, nil
|
|
}
|
|
|
|
// extractPercentage extracts the percentage (0.0-1.0) from source data
|
|
func extractPercentage(sourceFormat FormatGroup, sourceData map[string]interface{}) float64 {
|
|
switch sourceFormat {
|
|
case FormatGroupReflowable:
|
|
if p, ok := sourceData["percentage"].(float64); ok {
|
|
return p
|
|
}
|
|
// Try to calculate from character offset
|
|
if char, ok := sourceData["character"].(int64); ok {
|
|
if total, ok := sourceData["total_characters"].(int64); ok && total > 0 {
|
|
return float64(char) / float64(total)
|
|
}
|
|
}
|
|
|
|
case FormatGroupFixedLayout, FormatGroupComicArchive, FormatGroupAudiobook:
|
|
if page, ok := sourceData["page"].(int); ok {
|
|
if total, ok := sourceData["total_pages"].(int); ok && total > 0 {
|
|
return float64(page) / float64(total)
|
|
}
|
|
}
|
|
// Try direct percentage
|
|
if p, ok := sourceData["percentage"].(float64); ok {
|
|
return p
|
|
}
|
|
}
|
|
|
|
return 0.0
|
|
}
|
|
|
|
// PageToPercentage converts page/total_pages to percentage
|
|
func PageToPercentage(page, totalPages int) float64 {
|
|
if totalPages <= 0 {
|
|
return 0.0
|
|
}
|
|
percentage := float64(page) / float64(totalPages)
|
|
if percentage > 1.0 {
|
|
percentage = 1.0
|
|
}
|
|
if percentage < 0.0 {
|
|
percentage = 0.0
|
|
}
|
|
return percentage
|
|
}
|
|
|
|
// PercentageToPage converts percentage to page number
|
|
func PercentageToPage(percentage float64, totalPages int) int {
|
|
if percentage < 0.0 {
|
|
percentage = 0.0
|
|
}
|
|
if percentage > 1.0 {
|
|
percentage = 1.0
|
|
}
|
|
page := int(math.Round(float64(totalPages) * percentage))
|
|
if page < 0 {
|
|
page = 0
|
|
}
|
|
if page > totalPages {
|
|
page = totalPages
|
|
}
|
|
return page
|
|
}
|
|
|
|
// CharacterToPercentage converts character offset to percentage
|
|
func CharacterToPercentage(character, totalCharacters int64) float64 {
|
|
if totalCharacters <= 0 {
|
|
return 0.0
|
|
}
|
|
percentage := float64(character) / float64(totalCharacters)
|
|
if percentage > 1.0 {
|
|
percentage = 1.0
|
|
}
|
|
if percentage < 0.0 {
|
|
percentage = 0.0
|
|
}
|
|
return percentage
|
|
}
|
|
|
|
// PercentageToCharacter converts percentage to character offset
|
|
func PercentageToCharacter(percentage float64, totalCharacters int64) int64 {
|
|
if percentage < 0.0 {
|
|
percentage = 0.0
|
|
}
|
|
if percentage > 1.0 {
|
|
percentage = 1.0
|
|
}
|
|
char := int64(math.Round(float64(totalCharacters) * percentage))
|
|
if char < 0 {
|
|
char = 0
|
|
}
|
|
if char > totalCharacters {
|
|
char = totalCharacters
|
|
}
|
|
return char
|
|
}
|
|
|
|
// MergeProgress merges progress from two sources using "max progress wins" strategy
|
|
func MergeProgress(progressA, progressB map[string]interface{}) map[string]interface{} {
|
|
percA := extractPercentage(FormatGroupReflowable, progressA)
|
|
percB := extractPercentage(FormatGroupReflowable, progressB)
|
|
|
|
winner := progressB
|
|
if percA > percB {
|
|
winner = progressA
|
|
}
|
|
|
|
result := make(map[string]interface{})
|
|
for k, v := range winner {
|
|
result[k] = v
|
|
}
|
|
|
|
sources := []string{}
|
|
if srcA, ok := progressA["source"].(string); ok {
|
|
sources = append(sources, srcA)
|
|
}
|
|
if srcB, ok := progressB["source"].(string); ok {
|
|
sources = append(sources, srcB)
|
|
}
|
|
result["merged_from"] = sources
|
|
result["merge_timestamp"] = "now"
|
|
|
|
return result
|
|
}
|
|
|
|
// FormatProgressForDisplay formats progress for display based on format group
|
|
func FormatProgressForDisplay(formatGroup FormatGroup, progress map[string]interface{}) string {
|
|
percentage := extractPercentage(formatGroup, progress)
|
|
|
|
switch formatGroup {
|
|
case FormatGroupReflowable:
|
|
if chapter, ok := progress["chapter"].(int); ok {
|
|
return fmt.Sprintf("%.1f%% (Chapter %d)", percentage*100, chapter)
|
|
}
|
|
return fmt.Sprintf("%.1f%%", percentage*100)
|
|
|
|
case FormatGroupFixedLayout:
|
|
if page, ok := progress["page"].(int); ok {
|
|
if total, ok := progress["total_pages"].(int); ok {
|
|
return fmt.Sprintf("Page %d of %d (%.1f%%)", page, total, percentage*100)
|
|
}
|
|
return fmt.Sprintf("Page %d (%.1f%%)", page, percentage*100)
|
|
}
|
|
return fmt.Sprintf("%.1f%%", percentage*100)
|
|
|
|
case FormatGroupComicArchive:
|
|
if page, ok := progress["page"].(int); ok {
|
|
if total, ok := progress["total_pages"].(int); ok {
|
|
return fmt.Sprintf("Page %d of %d (%.0f%%)", page, total, percentage*100)
|
|
}
|
|
return fmt.Sprintf("Page %d (%.0f%%)", page, percentage*100)
|
|
}
|
|
return fmt.Sprintf("%.0f%%", percentage*100)
|
|
|
|
default:
|
|
return fmt.Sprintf("%.1f%%", percentage*100)
|
|
}
|
|
}
|
|
|
|
const CharsPerPage int64 = 1800
|
|
|
|
func EstimatedPages(totalCharacters int64) int {
|
|
if totalCharacters <= 0 {
|
|
return 0
|
|
}
|
|
return int(totalCharacters / CharsPerPage)
|
|
}
|
|
|
|
type ProgressService struct {
|
|
db *database.Queries
|
|
connManager *ConnectionManager
|
|
mediaPaths MediaPathResolver
|
|
}
|
|
|
|
// MediaPathResolver resolves a media item's library-relative file path to
|
|
// an absolute path, so the progress service can verify submitted anchors
|
|
// against the book itself.
|
|
type MediaPathResolver interface {
|
|
ResolveMediaPath(ctx context.Context, libraryID pgtype.UUID, relativePath string) (string, error)
|
|
}
|
|
|
|
func (s *ProgressService) SetMediaPathResolver(r MediaPathResolver) { s.mediaPaths = r }
|
|
|
|
func NewProgressService(db *database.Queries, connManager *ConnectionManager) *ProgressService {
|
|
return &ProgressService{db: db, connManager: connManager}
|
|
}
|
|
|
|
// SaveProgressRequest is the CANONICAL progress write payload — the web
|
|
// PUT binds it directly (identity fields are json:"-": server-owned, never
|
|
// client-settable) and every internal writer fills it programmatically.
|
|
// Wire keys are frozen: only additive json names are allowed here.
|
|
type SaveProgressRequest struct {
|
|
MediaItemID pgtype.UUID `json:"-"`
|
|
UserID pgtype.UUID `json:"-"`
|
|
Source string `json:"-"`
|
|
DeviceID pgtype.UUID `json:"-"`
|
|
|
|
Percentage *float64 `json:"percentage"`
|
|
Epubcfi *string `json:"epubcfi"`
|
|
ContextText *string `json:"context_text"`
|
|
CharacterOffset *int64 `json:"character_offset"`
|
|
Chapter *int `json:"chapter"`
|
|
ChapterProgress *float64 `json:"chapter_progress"`
|
|
CurrentPage *int `json:"current_page"`
|
|
TotalPages *int `json:"total_pages"`
|
|
ViewportX *float64 `json:"viewport_x"`
|
|
ViewportY *float64 `json:"viewport_y"`
|
|
ZoomLevel *float64 `json:"zoom_level"`
|
|
ScrollX *float64 `json:"scroll_position_x"`
|
|
ScrollY *float64 `json:"scroll_position_y"`
|
|
PanelNumber *int `json:"panel_number"`
|
|
ReadingMode *string `json:"reading_mode"`
|
|
// Optional client label ("android"/"ios") — the web handler normalizes
|
|
// it into Source (writer attribution) unless a paired device token
|
|
// claimed the push. Absent/unknown stays "web".
|
|
Origin string `json:"origin"`
|
|
// Audiobook locator (seconds into the current chapter)
|
|
AudioOffsetSeconds *int `json:"audio_offset_seconds"`
|
|
|
|
// AudioElapsedSeconds: the player's honest wall-clock PLAYING time
|
|
// since the last push (audio pushes fire while playing; pauses don't
|
|
// count). The capture prefers it over the inter-push delta, capped to
|
|
// the same 2h attributed ceiling.
|
|
AudioElapsedSeconds *int32 `json:"elapsed_seconds"`
|
|
|
|
DeviceType string `json:"-"`
|
|
DeviceName string `json:"-"`
|
|
Broadcast bool `json:"-"`
|
|
|
|
// ClearLocator declares that this write INTENDS to carry no position
|
|
// locator: percentage-only writers (devices whose converter declined,
|
|
// Kobo markup percentage syncs, queue entries without CFI). Without it,
|
|
// seeding below would re-pair the new percentage with the PREVIOUS
|
|
// writer's stale CFI/context — the incoherent row shape that feeds the
|
|
// client-side far-anchor restore floor. Locater-less rows are honest:
|
|
// percentage restores are approximate, and the next anchor-bearing
|
|
// write re-establishes the pair. Request-supplied fields always win.
|
|
ClearLocator bool `json:"-"`
|
|
|
|
// SkipConflictDetection declares this write APPLIES a conflict
|
|
// resolution rather than continuing the sync stream: the two flows
|
|
// (single + bulk) apply the winner BEFORE the conflict row is marked
|
|
// resolved, so the funnel's LWW detector would see the losing source
|
|
// within the 5-minute window and file a duplicate pending conflict on
|
|
// the item the user just resolved. Server-owned (json:"-").
|
|
SkipConflictDetection bool `json:"-"`
|
|
}
|
|
|
|
func (s *ProgressService) SaveProgress(ctx context.Context, req SaveProgressRequest) (database.ReadingProgress, error) {
|
|
if req.Source == "" {
|
|
req.Source = "unknown"
|
|
}
|
|
|
|
mediaItem, err := s.db.GetMediaItem(ctx, req.MediaItemID)
|
|
if err != nil {
|
|
return database.ReadingProgress{}, fmt.Errorf("media item not found: %w", err)
|
|
}
|
|
|
|
existing, err := s.db.GetReadingProgress(ctx, database.GetReadingProgressParams{
|
|
MediaItemID: req.MediaItemID,
|
|
UserID: req.UserID,
|
|
})
|
|
hasExisting := err == nil
|
|
if err != nil && err != pgx.ErrNoRows {
|
|
return database.ReadingProgress{}, fmt.Errorf("failed to read existing progress: %w", err)
|
|
}
|
|
|
|
params := database.UpdateUniversalProgressParams{
|
|
MediaItemID: req.MediaItemID,
|
|
UserID: req.UserID,
|
|
}
|
|
|
|
if hasExisting {
|
|
params.Percentage = existing.Percentage
|
|
params.CharacterOffset = existing.CharacterOffset
|
|
params.Epubcfi = existing.Epubcfi
|
|
params.ContextText = existing.ContextText
|
|
params.Chapter = existing.Chapter
|
|
params.ChapterProgress = existing.ChapterProgress
|
|
params.AudioOffsetSeconds = existing.AudioOffsetSeconds
|
|
params.ViewportX = existing.ViewportX
|
|
params.ViewportY = existing.ViewportY
|
|
params.ZoomLevel = existing.ZoomLevel
|
|
params.ScrollPositionX = existing.ScrollPositionX
|
|
params.ScrollPositionY = existing.ScrollPositionY
|
|
params.PanelNumber = existing.PanelNumber
|
|
params.ReadingMode = existing.ReadingMode
|
|
params.CurrentPage = existing.CurrentPage
|
|
params.TotalPages = existing.TotalPages
|
|
params.LastSyncDevice = existing.LastSyncDevice
|
|
params.LastSyncSource = existing.LastSyncSource
|
|
}
|
|
|
|
if req.Percentage != nil {
|
|
params.Percentage = pgtype.Float8{Float64: *req.Percentage, Valid: true}
|
|
}
|
|
if req.Epubcfi != nil {
|
|
params.Epubcfi = pgtype.Text{String: *req.Epubcfi, Valid: *req.Epubcfi != ""}
|
|
}
|
|
if req.ContextText != nil {
|
|
params.ContextText = pgtype.Text{String: *req.ContextText, Valid: *req.ContextText != ""}
|
|
}
|
|
if req.CharacterOffset != nil {
|
|
params.CharacterOffset = pgtype.Int8{Int64: *req.CharacterOffset, Valid: true}
|
|
}
|
|
if req.Chapter != nil {
|
|
params.Chapter = pgtype.Int4{Int32: int32(*req.Chapter), Valid: true}
|
|
}
|
|
if req.ChapterProgress != nil {
|
|
params.ChapterProgress = pgtype.Float8{Float64: *req.ChapterProgress, Valid: true}
|
|
}
|
|
if req.AudioOffsetSeconds != nil {
|
|
params.AudioOffsetSeconds = pgtype.Int4{Int32: int32(*req.AudioOffsetSeconds), Valid: true}
|
|
}
|
|
if req.CurrentPage != nil {
|
|
params.CurrentPage = pgtype.Int4{Int32: int32(*req.CurrentPage), Valid: true}
|
|
}
|
|
if req.TotalPages != nil {
|
|
params.TotalPages = pgtype.Int4{Int32: int32(*req.TotalPages), Valid: true}
|
|
}
|
|
if req.ViewportX != nil {
|
|
params.ViewportX = pgtype.Float8{Float64: *req.ViewportX, Valid: true}
|
|
}
|
|
if req.ViewportY != nil {
|
|
params.ViewportY = pgtype.Float8{Float64: *req.ViewportY, Valid: true}
|
|
}
|
|
if req.ZoomLevel != nil {
|
|
params.ZoomLevel = pgtype.Float8{Float64: *req.ZoomLevel, Valid: true}
|
|
}
|
|
if req.ScrollX != nil {
|
|
params.ScrollPositionX = pgtype.Float8{Float64: *req.ScrollX, Valid: true}
|
|
}
|
|
if req.ScrollY != nil {
|
|
params.ScrollPositionY = pgtype.Float8{Float64: *req.ScrollY, Valid: true}
|
|
}
|
|
if req.PanelNumber != nil {
|
|
params.PanelNumber = pgtype.Int4{Int32: int32(*req.PanelNumber), Valid: true}
|
|
}
|
|
if req.ReadingMode != nil {
|
|
params.ReadingMode = pgtype.Text{String: *req.ReadingMode, Valid: true}
|
|
}
|
|
|
|
// Locator coherence: a write that declares no locator must not
|
|
// silently re-pair its new percentage with the previous writer's
|
|
// stale CFI/context (inherited by the seeding above). Clear them so
|
|
// the row is an honest percentage-only position. Request-carried
|
|
// fields always win — and a request-carried context survives so the
|
|
// verify/heal pass below can mint a fresh anchor from it.
|
|
if req.ClearLocator && req.Epubcfi == nil {
|
|
params.Epubcfi = pgtype.Text{}
|
|
if req.ContextText == nil {
|
|
params.ContextText = pgtype.Text{}
|
|
}
|
|
if req.CharacterOffset == nil {
|
|
params.CharacterOffset = pgtype.Int8{}
|
|
}
|
|
}
|
|
|
|
params.LastSyncDevice = pgtype.Text{String: req.Source, Valid: true}
|
|
params.LastSyncSource = pgtype.Text{String: req.Source, Valid: true}
|
|
|
|
formatGroup := FormatGroup(mediaItem.FormatGroup)
|
|
isFixed := formatGroup == FormatGroupFixedLayout || formatGroup == FormatGroupComicArchive
|
|
if isFixed {
|
|
// Fixed-layout & comic formats: the page index is the canonical locator.
|
|
// CFI/character-offset are meaningless for image-based content, so clear
|
|
// any stale value that may have been stored by the web reader (which
|
|
// generates fake CFIs for comics).
|
|
params.Epubcfi = pgtype.Text{}
|
|
params.CharacterOffset = pgtype.Int8{}
|
|
if params.CurrentPage.Valid && params.TotalPages.Valid && params.TotalPages.Int32 > 0 {
|
|
pct := PageToPercentage(int(params.CurrentPage.Int32), int(params.TotalPages.Int32))
|
|
params.Percentage = pgtype.Float8{Float64: pct, Valid: true}
|
|
} else if params.Percentage.Valid && params.TotalPages.Valid && params.TotalPages.Int32 > 0 && !params.CurrentPage.Valid {
|
|
page := PercentageToPage(params.Percentage.Float64, int(params.TotalPages.Int32))
|
|
params.CurrentPage = pgtype.Int4{Int32: int32(page), Valid: true}
|
|
}
|
|
} else {
|
|
if params.Percentage.Valid && !params.CharacterOffset.Valid && mediaItem.TotalCharacters.Valid && mediaItem.TotalCharacters.Int64 > 0 {
|
|
charOff := PercentageToCharacter(params.Percentage.Float64, mediaItem.TotalCharacters.Int64)
|
|
params.CharacterOffset = pgtype.Int8{Int64: charOff, Valid: true}
|
|
}
|
|
if params.Percentage.Valid && !params.CurrentPage.Valid && params.TotalPages.Valid && params.TotalPages.Int32 > 0 {
|
|
page := PercentageToPage(params.Percentage.Float64, int(params.TotalPages.Int32))
|
|
params.CurrentPage = pgtype.Int4{Int32: int32(page), Valid: true}
|
|
}
|
|
if params.CurrentPage.Valid && params.TotalPages.Valid && params.TotalPages.Int32 > 0 && !params.Percentage.Valid {
|
|
pct := PageToPercentage(int(params.CurrentPage.Int32), int(params.TotalPages.Int32))
|
|
params.Percentage = pgtype.Float8{Float64: pct, Valid: true}
|
|
}
|
|
if params.CharacterOffset.Valid && mediaItem.TotalCharacters.Valid && mediaItem.TotalCharacters.Int64 > 0 && !params.Percentage.Valid {
|
|
pct := CharacterToPercentage(params.CharacterOffset.Int64, mediaItem.TotalCharacters.Int64)
|
|
params.Percentage = pgtype.Float8{Float64: pct, Valid: true}
|
|
}
|
|
}
|
|
|
|
// Canonical position authority: resolve the submitted CFI against the
|
|
// book itself, cross-check the submitted context text, and heal the
|
|
// anchor by text search on any mismatch. Clients submit what their
|
|
// renderer can reliably observe (percentage + visible text); the
|
|
// server owns the structural math and keeps buggy clients from
|
|
// poisoning stored positions.
|
|
if !isFixed && s.mediaPaths != nil && params.ContextText.Valid &&
|
|
isConvertible(string(formatGroup)) && formatGroup != "" {
|
|
if path, perr := s.mediaPaths.ResolveMediaPath(ctx, mediaItem.LibraryID, mediaItem.FilePath); perr == nil && path != "" {
|
|
submittedCFI := ""
|
|
if params.Epubcfi.Valid {
|
|
submittedCFI = params.Epubcfi.String
|
|
}
|
|
submittedPct := 0.0
|
|
if params.Percentage.Valid {
|
|
submittedPct = params.Percentage.Float64
|
|
}
|
|
if finalCFI, _, _, _, bookOff, healedPct, healed, verr := VerifyProgressAnchor(path, submittedCFI, params.ContextText.String, submittedPct); verr != nil {
|
|
log.Printf("Bookhoard: progress anchor verify failed for %s: %v", req.MediaItemID.String(), verr)
|
|
} else {
|
|
if healed || (!params.Epubcfi.Valid && finalCFI != "") {
|
|
params.Epubcfi = pgtype.Text{String: finalCFI, Valid: true}
|
|
if healedPct != nil {
|
|
params.Percentage = pgtype.Float8{Float64: *healedPct, Valid: true}
|
|
}
|
|
}
|
|
// The verified book-wide offset refreshes the column on
|
|
// EVERY verified save — not only heals — so it never goes
|
|
// stale behind the anchor (and a block-relative offset is
|
|
// never written into the book-level column).
|
|
if bookOff != nil {
|
|
params.CharacterOffset = pgtype.Int8{Int64: int64(*bookOff), Valid: true}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
conflictDetected := false
|
|
if !req.SkipConflictDetection && hasExisting && existing.LastSyncSource.Valid && existing.LastSyncSource.String != req.Source {
|
|
if existing.LastSyncTimestamp.Valid {
|
|
timeDiff := time.Since(existing.LastSyncTimestamp.Time)
|
|
if timeDiff < 5*time.Minute {
|
|
existingPct := 0.0
|
|
if existing.Percentage.Valid {
|
|
existingPct = existing.Percentage.Float64
|
|
}
|
|
newPct := 0.0
|
|
if params.Percentage.Valid {
|
|
newPct = params.Percentage.Float64
|
|
}
|
|
diff := newPct - existingPct
|
|
if diff < 0 {
|
|
diff = -diff
|
|
}
|
|
if diff > 0.01 {
|
|
recentlyResolved, _ := s.db.HasRecentConflictResolution(ctx, database.HasRecentConflictResolutionParams{
|
|
MediaItemID: req.MediaItemID,
|
|
UserID: req.UserID,
|
|
})
|
|
if !recentlyResolved {
|
|
conflictDetected = true
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
result, err := s.db.UpdateUniversalProgress(ctx, params)
|
|
if err != nil {
|
|
return database.ReadingProgress{}, fmt.Errorf("failed to upsert progress: %w", err)
|
|
}
|
|
|
|
if conflictDetected {
|
|
conflictData := map[string]interface{}{
|
|
req.Source: map[string]interface{}{
|
|
"source": req.Source,
|
|
"timestamp": time.Now().Format(time.RFC3339),
|
|
"data": buildProgressSnapshot(req),
|
|
},
|
|
existing.LastSyncSource.String: map[string]interface{}{
|
|
"source": existing.LastSyncSource.String,
|
|
"timestamp": existing.LastSyncTimestamp.Time.Format(time.RFC3339),
|
|
"data": map[string]interface{}{
|
|
"percentage": float64Ptr(existing.Percentage),
|
|
"epubcfi": textPtr(existing.Epubcfi),
|
|
"chapter": int32Ptr(existing.Chapter),
|
|
"character": int64Ptr(existing.CharacterOffset),
|
|
"page": int32Ptr(existing.CurrentPage),
|
|
"total_pages": int32Ptr(existing.TotalPages),
|
|
},
|
|
},
|
|
}
|
|
conflictJSON, _ := json.Marshal(conflictData)
|
|
_, err := s.db.CreateSyncConflict(ctx, database.CreateSyncConflictParams{
|
|
MediaItemID: req.MediaItemID,
|
|
UserID: req.UserID,
|
|
ConflictType: "progress",
|
|
ConflictData: conflictJSON,
|
|
})
|
|
if err != nil {
|
|
log.Printf("Failed to record sync conflict: %v", err)
|
|
} else if s.connManager != nil {
|
|
s.connManager.BroadcastConflictNotification(
|
|
uuid.UUID(req.UserID.Bytes).String(),
|
|
req.MediaItemID.Bytes, "detection", "")
|
|
}
|
|
}
|
|
|
|
if s.connManager != nil && req.Broadcast {
|
|
deviceName := req.DeviceName
|
|
if deviceName == "" {
|
|
deviceName = req.Source + " Device"
|
|
}
|
|
pct := 0.0
|
|
if result.Percentage.Valid {
|
|
pct = result.Percentage.Float64
|
|
}
|
|
s.connManager.BroadcastProgressUpdate(
|
|
uuid.UUID(req.UserID.Bytes).String(),
|
|
uuid.UUID(req.MediaItemID.Bytes),
|
|
pct,
|
|
SourceDevice{
|
|
ID: uuid.UUID(req.DeviceID.Bytes).String(),
|
|
Name: deviceName,
|
|
Type: req.Source,
|
|
},
|
|
)
|
|
}
|
|
|
|
// Reading-history capture: threading stats off the sync stream. Every
|
|
// successful progress write yields one reading_history session row —
|
|
// the span since the PREVIOUS write instant (bounds are one or the
|
|
// other: the row is recorded regardless, time attribution is
|
|
// conservative). Client cadence drives the granularity (web/app
|
|
// debounced pushes, KOReader page-count pushes); nothing second-level
|
|
// is promised and nothing client-side changes.
|
|
s.captureReadingSession(ctx, req, existing, hasExisting, result)
|
|
|
|
return result, nil
|
|
}
|
|
|
|
// maxAttributedSessionSeconds caps a derived session: pushes farther
|
|
// apart than this carry unattributable idle time (overnight suspend →
|
|
// resume, offline queue flushes) and are kept as zero-second markers
|
|
// instead of inflating totals.
|
|
const maxAttributedSessionSeconds = int32(2 * 60 * 60)
|
|
|
|
// captureReadingSession records one reading_history row for this push.
|
|
// Never fails the sync: capture errors are logged and swallowed.
|
|
func (s *ProgressService) captureReadingSession(ctx context.Context, req SaveProgressRequest, existing database.ReadingProgress, hasExisting bool, saved database.ReadingProgress) {
|
|
var seconds int32
|
|
sessionEnd := saved.LastReadAt
|
|
sessionStart := sessionEnd // first-contact writes start at the push itself
|
|
if sessionEnd.Valid && hasExisting && existing.LastReadAt.Valid {
|
|
sessionStart = existing.LastReadAt
|
|
if elapsed := int32(sessionEnd.Time.Sub(existing.LastReadAt.Time).Seconds()); elapsed > 0 {
|
|
if elapsed <= maxAttributedSessionSeconds {
|
|
seconds = elapsed // verified-continuous span
|
|
} // else: idle gap — marker row, zero time
|
|
}
|
|
// Wall-clock override: honest player-reported playing time beats
|
|
// the inter-push delta (pauses shouldn't count as reading). Same
|
|
// 2h ceiling applies.
|
|
if req.AudioElapsedSeconds != nil && *req.AudioElapsedSeconds > 0 && *req.AudioElapsedSeconds <= maxAttributedSessionSeconds {
|
|
seconds = *req.AudioElapsedSeconds
|
|
}
|
|
}
|
|
|
|
// Pages read: only when both sides carry a real page index (0 means
|
|
// unknown for several clients, e.g. LOC-less KOReader pushes).
|
|
var pages pgtype.Int4
|
|
if hasExisting && existing.CurrentPage.Valid && existing.CurrentPage.Int32 > 0 &&
|
|
saved.CurrentPage.Valid && saved.CurrentPage.Int32 > 0 {
|
|
if deltaPages := saved.CurrentPage.Int32 - existing.CurrentPage.Int32; deltaPages > 0 && deltaPages <= 5000 {
|
|
pages = pgtype.Int4{Int32: deltaPages, Valid: true}
|
|
}
|
|
}
|
|
|
|
// device_id references real devices (FK). The web/app progress paths
|
|
// historically stuff the USER id into DeviceID (it only fed a
|
|
// broadcast label before stats existed) — probe before writing,
|
|
// otherwise capture a device-less session and keep the label in
|
|
// metadata only.
|
|
var deviceID pgtype.UUID
|
|
if req.DeviceID.Valid {
|
|
exists, probeErr := s.db.DeviceExistsForUser(ctx, database.DeviceExistsForUserParams{
|
|
DeviceID: req.DeviceID,
|
|
UserID: req.UserID,
|
|
})
|
|
if probeErr == nil && exists {
|
|
deviceID = req.DeviceID
|
|
}
|
|
}
|
|
metadata := map[string]string{"source": req.Source}
|
|
|
|
_, err := s.db.CreateReadingHistory(ctx, database.CreateReadingHistoryParams{
|
|
UserID: req.UserID,
|
|
MediaItemID: req.MediaItemID,
|
|
DeviceID: deviceID,
|
|
ProgressPercentage: saved.Percentage,
|
|
ReadingSessionStart: sessionStart,
|
|
ReadingSessionEnd: sessionEnd,
|
|
PagesRead: pages,
|
|
TimeSpentSeconds: pgtype.Int4{Int32: seconds, Valid: true},
|
|
DeviceMetadata: mustJSON(metadata),
|
|
})
|
|
if err != nil {
|
|
log.Printf("reading-history capture failed (non-fatal) for %s: %v", uniqueKeyString(req), err)
|
|
}
|
|
}
|
|
|
|
func uniqueKeyString(req SaveProgressRequest) string {
|
|
return uuid.UUID(req.MediaItemID.Bytes).String()
|
|
}
|
|
|
|
// mustJSON marshals metadata or an empty document on failure — capture
|
|
// must never break on a serialization corner.
|
|
func mustJSON(v interface{}) []byte {
|
|
data, err := json.Marshal(v)
|
|
if err != nil {
|
|
return []byte(`{}`)
|
|
}
|
|
return data
|
|
}
|
|
|
|
func buildProgressSnapshot(req SaveProgressRequest) map[string]interface{} {
|
|
data := map[string]interface{}{}
|
|
if req.Percentage != nil {
|
|
data["percentage"] = *req.Percentage
|
|
}
|
|
if req.Epubcfi != nil {
|
|
data["epubcfi"] = *req.Epubcfi
|
|
}
|
|
if req.ContextText != nil {
|
|
data["context_text"] = *req.ContextText
|
|
}
|
|
if req.Chapter != nil {
|
|
data["chapter"] = *req.Chapter
|
|
}
|
|
if req.CharacterOffset != nil {
|
|
data["character"] = *req.CharacterOffset
|
|
}
|
|
if req.CurrentPage != nil {
|
|
data["page"] = *req.CurrentPage
|
|
}
|
|
if req.TotalPages != nil {
|
|
data["total_pages"] = *req.TotalPages
|
|
}
|
|
return data
|
|
}
|
|
|
|
func float64Ptr(v pgtype.Float8) *float64 {
|
|
if v.Valid {
|
|
return &v.Float64
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func textPtr(v pgtype.Text) *string {
|
|
if v.Valid {
|
|
return &v.String
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func int32Ptr(v pgtype.Int4) *int {
|
|
if v.Valid {
|
|
val := int(v.Int32)
|
|
return &val
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func int64Ptr(v pgtype.Int8) *int64 {
|
|
if v.Valid {
|
|
return &v.Int64
|
|
}
|
|
return nil
|
|
}
|