Alexander
5a5660bf21
Ultraworked with [Sisyphus](https://github.com/code-yeongyu/claude-agent) Co-authored-by: Sisyphus <clio-agent@sisyphuslabs.ai>
432 lines
13 KiB
Go
432 lines
13 KiB
Go
package internal
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"sync"
|
|
"time"
|
|
|
|
"github.com/rs/zerolog/log"
|
|
"google.golang.org/grpc/codes"
|
|
"google.golang.org/grpc/status"
|
|
|
|
pb "homelab.lan/music-agregator/gen/music_agregator/v1"
|
|
"homelab.lan/music-agregator/internal/database"
|
|
)
|
|
|
|
var MaxPromptTimeout = 300 * time.Second
|
|
|
|
type monitorWorkflow struct {
|
|
mode pb.InteractionMode
|
|
req *pb.StartMonitorRequest
|
|
service *MusicAgregatorService
|
|
publisher EventPublisher
|
|
|
|
stream pb.MusicAgregatorService_MonitorAlbumStreamServer
|
|
decisions chan *pb.UserDecision
|
|
cancel context.CancelFunc
|
|
|
|
addedHash string
|
|
workflowRunID string
|
|
|
|
mu sync.Mutex
|
|
promptID int
|
|
}
|
|
|
|
func (w *monitorWorkflow) nextPromptID() string {
|
|
w.mu.Lock()
|
|
defer w.mu.Unlock()
|
|
w.promptID++
|
|
return fmt.Sprintf("prompt-%d", w.promptID)
|
|
}
|
|
|
|
func (w *monitorWorkflow) promptAndWait(ctx context.Context, prompt *pb.PromptForDecision) (*pb.UserDecision, error) {
|
|
if w.mode == pb.InteractionMode_INTERACTION_MODE_AUTOMATIC {
|
|
return w.defaultDecision(prompt), nil
|
|
}
|
|
|
|
if w.stream == nil {
|
|
return w.defaultDecision(prompt), nil
|
|
}
|
|
|
|
if err := w.stream.Send(&pb.MonitorAlbumStreamResponse{
|
|
Message: &pb.MonitorAlbumStreamResponse_Prompt{Prompt: prompt},
|
|
}); err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
timeout := time.Duration(prompt.TimeoutSeconds) * time.Second
|
|
if timeout == 0 || timeout > MaxPromptTimeout {
|
|
timeout = MaxPromptTimeout
|
|
}
|
|
|
|
timeoutCtx, cancel := context.WithTimeout(ctx, timeout)
|
|
defer cancel()
|
|
|
|
select {
|
|
case decision := <-w.decisions:
|
|
if decision.PromptId != prompt.PromptId {
|
|
return nil, status.Error(codes.InvalidArgument, "prompt_id mismatch")
|
|
}
|
|
return decision, nil
|
|
case <-timeoutCtx.Done():
|
|
return w.defaultDecision(prompt), nil
|
|
}
|
|
}
|
|
|
|
func (w *monitorWorkflow) defaultDecision(prompt *pb.PromptForDecision) *pb.UserDecision {
|
|
decision := &pb.UserDecision{PromptId: prompt.PromptId}
|
|
|
|
switch prompt.Type {
|
|
case pb.PromptType_PROMPT_TYPE_CONFIRM:
|
|
decision.Decision = &pb.UserDecision_Confirm{
|
|
Confirm: prompt.GetConfirm().GetDefaultValue(),
|
|
}
|
|
case pb.PromptType_PROMPT_TYPE_SELECT_ONE:
|
|
decision.Decision = &pb.UserDecision_SelectedId{
|
|
SelectedId: prompt.GetSelectOne().GetDefaultId(),
|
|
}
|
|
case pb.PromptType_PROMPT_TYPE_SELECT_MANY:
|
|
decision.Decision = &pb.UserDecision_SelectedIds{
|
|
SelectedIds: &pb.SelectedIds{Ids: prompt.GetSelectMany().GetDefaultIds()},
|
|
}
|
|
}
|
|
|
|
return decision
|
|
}
|
|
|
|
func (w *monitorWorkflow) receiveDecisions(ctx context.Context) {
|
|
if w.stream == nil {
|
|
return
|
|
}
|
|
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
default:
|
|
}
|
|
|
|
msg, err := w.stream.Recv()
|
|
if err != nil {
|
|
return
|
|
}
|
|
|
|
if msg.GetCancel() != nil {
|
|
if w.cancel != nil {
|
|
w.cancel()
|
|
}
|
|
return
|
|
}
|
|
|
|
if decision := msg.GetDecision(); decision != nil {
|
|
select {
|
|
case w.decisions <- decision:
|
|
default:
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
func (w *monitorWorkflow) run(ctx context.Context) error {
|
|
select {
|
|
case <-ctx.Done():
|
|
return ctx.Err()
|
|
default:
|
|
}
|
|
|
|
w.publisher.PublishStatus(ctx, pb.MonitorStep_MONITOR_STEP_FETCHING_METADATA, "Fetching album metadata...", nil)
|
|
|
|
album, err := w.service.metadata.GetAlbum(ctx, w.req.AlbumId)
|
|
if err != nil {
|
|
if ctx.Err() != nil {
|
|
return ctx.Err()
|
|
}
|
|
log.Error().Err(err).Str("album_id", w.req.AlbumId).Msg("failed to get album")
|
|
w.publisher.PublishError(ctx, pb.MonitorStep_MONITOR_STEP_FETCHING_METADATA, err, false)
|
|
return err
|
|
}
|
|
|
|
artistName := ""
|
|
if len(album.GetArtists()) > 0 {
|
|
artistName = album.GetArtists()[0].GetArtist().GetName()
|
|
}
|
|
|
|
w.publisher.PublishStatus(ctx, pb.MonitorStep_MONITOR_STEP_FETCHING_METADATA,
|
|
fmt.Sprintf("Got metadata: %s - %s", artistName, album.GetTitle()),
|
|
&pb.StreamAlbumInfo{
|
|
Artist: artistName,
|
|
Title: album.GetTitle(),
|
|
ReleaseDate: album.GetReleaseDate(),
|
|
})
|
|
|
|
w.publisher.PublishStatus(ctx, pb.MonitorStep_MONITOR_STEP_CHECKING_OWNED, "Checking if already owned...", nil)
|
|
|
|
dbAlbum, _ := w.service.metadata.GetAlbumByExternalID(ctx, album.GetId())
|
|
if dbAlbum != nil {
|
|
w.publisher.SetAlbumID(dbAlbum.ID)
|
|
w.service.metadata.SetAlbumMonitorState(ctx, dbAlbum.ID, database.Monitored)
|
|
dbAlbum.MonitorState = database.Monitored
|
|
|
|
if w.workflowRunID == "" {
|
|
run := &database.WorkflowRun{AlbumID: dbAlbum.ID, Quality: w.req.Quality.String()}
|
|
if err := w.service.workflowRuns.Create(ctx, run); err != nil && err != database.ErrWorkflowAlreadyRunning {
|
|
log.Warn().Err(err).Msg("failed to create workflow run")
|
|
} else if err == nil {
|
|
w.workflowRunID = run.ID
|
|
w.publisher.SetWorkflowRunID(run.ID)
|
|
}
|
|
}
|
|
|
|
qualityStr := normalizeQuality(w.req.Quality, 0, 0)
|
|
owned, err := w.service.downloads.HasAlbumInQuality(ctx, dbAlbum.ID, w.req.Quality.String(), qualityStr)
|
|
if err == nil && owned {
|
|
w.publisher.PublishStatus(ctx, pb.MonitorStep_MONITOR_STEP_CHECKING_OWNED,
|
|
fmt.Sprintf("Already owned in %s quality", qualityStr), nil)
|
|
|
|
if w.mode == pb.InteractionMode_INTERACTION_MODE_MANUAL {
|
|
decision, err := w.promptAndWait(ctx, &pb.PromptForDecision{
|
|
PromptId: w.nextPromptID(),
|
|
Type: pb.PromptType_PROMPT_TYPE_CONFIRM,
|
|
Message: "Album already owned. Download anyway?",
|
|
|
|
Options: &pb.PromptForDecision_Confirm{
|
|
Confirm: &pb.ConfirmPrompt{
|
|
ConfirmLabel: "Download anyway",
|
|
CancelLabel: "Skip",
|
|
DefaultValue: false,
|
|
},
|
|
},
|
|
})
|
|
if err != nil {
|
|
w.publisher.PublishError(ctx, pb.MonitorStep_MONITOR_STEP_CHECKING_OWNED, err, false)
|
|
return err
|
|
}
|
|
if !decision.GetConfirm() {
|
|
w.publisher.PublishStatus(ctx, pb.MonitorStep_MONITOR_STEP_COMPLETE, "Skipped - already owned", nil)
|
|
return w.publisher.PublishResult(ctx, w.service.buildMonitorAlbumResponse(ctx, album, dbAlbum, nil))
|
|
}
|
|
} else {
|
|
w.publisher.PublishStatus(ctx, pb.MonitorStep_MONITOR_STEP_COMPLETE, "Already owned", nil)
|
|
return w.publisher.PublishResult(ctx, w.service.buildMonitorAlbumResponse(ctx, album, dbAlbum, nil))
|
|
}
|
|
}
|
|
}
|
|
|
|
w.publisher.PublishStatus(ctx, pb.MonitorStep_MONITOR_STEP_SEARCHING_INDEXER,
|
|
fmt.Sprintf("Searching indexers for %s - %s...", artistName, album.GetTitle()), nil)
|
|
|
|
searchResult, err := w.service.searchIndexer(album, w.req.IndexerOptions.GetTracker())
|
|
if err != nil {
|
|
w.publisher.PublishError(ctx, pb.MonitorStep_MONITOR_STEP_SEARCHING_INDEXER, err, true)
|
|
return err
|
|
}
|
|
|
|
parsed := w.service.parseSearchResults(searchResult, album)
|
|
|
|
if len(parsed) > 0 {
|
|
summaries := make([]*pb.TorrentSummary, len(parsed))
|
|
for i, p := range parsed {
|
|
summaries[i] = &pb.TorrentSummary{
|
|
Id: fmt.Sprintf("torrent-%d", i),
|
|
Title: p.item.Title,
|
|
Tracker: p.item.Tracker,
|
|
Seeders: int32(p.item.Seeders),
|
|
Format: p.rel.Format.String(),
|
|
Lossless: p.rel.Format.IsLossless(),
|
|
}
|
|
}
|
|
w.publisher.PublishStatus(ctx, pb.MonitorStep_MONITOR_STEP_PARSING_RESULTS,
|
|
fmt.Sprintf("Parsed %d from %d torrents", len(parsed), len(searchResult.Items)),
|
|
&pb.TorrentList{Torrents: summaries})
|
|
} else {
|
|
w.publisher.PublishStatus(ctx, pb.MonitorStep_MONITOR_STEP_PARSING_RESULTS,
|
|
fmt.Sprintf("Found %d torrents, none parseable", len(searchResult.Items)), nil)
|
|
}
|
|
|
|
if len(parsed) == 0 {
|
|
w.publisher.PublishStatus(ctx, pb.MonitorStep_MONITOR_STEP_COMPLETE, "No parseable results found", nil)
|
|
return w.publisher.PublishResult(ctx, w.service.buildMonitorAlbumResponse(ctx, album, dbAlbum, nil))
|
|
}
|
|
|
|
if w.mode == pb.InteractionMode_INTERACTION_MODE_MANUAL && len(parsed) > 1 {
|
|
options := make([]*pb.SelectOption, len(parsed))
|
|
defaultIDs := make([]string, len(parsed))
|
|
for i, p := range parsed {
|
|
id := fmt.Sprintf("torrent-%d", i)
|
|
options[i] = &pb.SelectOption{
|
|
Id: id,
|
|
Label: p.item.Title,
|
|
Description: fmt.Sprintf("%s - %d seeders", p.item.Tracker, p.item.Seeders),
|
|
}
|
|
defaultIDs[i] = id
|
|
}
|
|
|
|
decision, err := w.promptAndWait(ctx, &pb.PromptForDecision{
|
|
PromptId: w.nextPromptID(),
|
|
Type: pb.PromptType_PROMPT_TYPE_SELECT_MANY,
|
|
Message: "Select torrents to consider",
|
|
|
|
Options: &pb.PromptForDecision_SelectMany{
|
|
SelectMany: &pb.SelectManyPrompt{
|
|
Options: options,
|
|
DefaultIds: defaultIDs,
|
|
MinSelections: 1,
|
|
MaxSelections: int32(len(parsed)),
|
|
},
|
|
},
|
|
})
|
|
if err != nil {
|
|
w.publisher.PublishError(ctx, pb.MonitorStep_MONITOR_STEP_PARSING_RESULTS, err, false)
|
|
return err
|
|
}
|
|
|
|
selectedIDs := make(map[string]bool)
|
|
if ids := decision.GetSelectedIds(); ids != nil {
|
|
for _, id := range ids.GetIds() {
|
|
selectedIDs[id] = true
|
|
}
|
|
}
|
|
|
|
var selected []parsedItem
|
|
for i, p := range parsed {
|
|
id := fmt.Sprintf("torrent-%d", i)
|
|
if selectedIDs[id] {
|
|
selected = append(selected, p)
|
|
}
|
|
}
|
|
if len(selected) > 0 {
|
|
parsed = selected
|
|
}
|
|
}
|
|
|
|
w.publisher.PublishStatus(ctx, pb.MonitorStep_MONITOR_STEP_FILTERING_QUALITY,
|
|
fmt.Sprintf("Filtering %d results by quality...", len(parsed)), nil)
|
|
|
|
filtered := filterByQuality(parsed, w.req.Quality)
|
|
if len(filtered) == 0 {
|
|
log.Warn().Str("album", album.GetTitle()).Str("quality", w.req.Quality.String()).Msg("no releases match quality filter")
|
|
w.publisher.PublishStatus(ctx, pb.MonitorStep_MONITOR_STEP_COMPLETE, "No releases match quality filter", nil)
|
|
return w.publisher.PublishResult(ctx, w.service.buildMonitorAlbumResponse(ctx, album, dbAlbum, nil))
|
|
}
|
|
|
|
var best parsedItem
|
|
if w.mode == pb.InteractionMode_INTERACTION_MODE_MANUAL && len(filtered) > 1 {
|
|
options := make([]*pb.SelectOption, len(filtered))
|
|
for i, p := range filtered {
|
|
options[i] = &pb.SelectOption{
|
|
Id: fmt.Sprintf("release-%d", i),
|
|
Label: p.item.Title,
|
|
Description: fmt.Sprintf("%s - %d seeders - %s", p.item.Tracker, p.item.Seeders, p.rel.Format.String()),
|
|
}
|
|
}
|
|
|
|
bestIdx := 0
|
|
for i, p := range filtered {
|
|
if p.item.Seeders > filtered[bestIdx].item.Seeders {
|
|
bestIdx = i
|
|
}
|
|
}
|
|
|
|
decision, err := w.promptAndWait(ctx, &pb.PromptForDecision{
|
|
PromptId: w.nextPromptID(),
|
|
Type: pb.PromptType_PROMPT_TYPE_SELECT_ONE,
|
|
Message: "Select release",
|
|
|
|
Options: &pb.PromptForDecision_SelectOne{
|
|
SelectOne: &pb.SelectOnePrompt{
|
|
Options: options,
|
|
DefaultId: fmt.Sprintf("release-%d", bestIdx),
|
|
},
|
|
},
|
|
})
|
|
if err != nil {
|
|
w.publisher.PublishError(ctx, pb.MonitorStep_MONITOR_STEP_SELECTING_RELEASE, err, false)
|
|
return err
|
|
}
|
|
|
|
selectedIdx := 0
|
|
if id := decision.GetSelectedId(); id != "" {
|
|
for i := range filtered {
|
|
if fmt.Sprintf("release-%d", i) == id {
|
|
selectedIdx = i
|
|
break
|
|
}
|
|
}
|
|
}
|
|
best = filtered[selectedIdx]
|
|
} else {
|
|
best = selectBestRelease(filtered)
|
|
}
|
|
|
|
w.publisher.PublishStatus(ctx, pb.MonitorStep_MONITOR_STEP_SELECTING_RELEASE,
|
|
fmt.Sprintf("Selected: %s (%d seeders)", best.item.Title, best.item.Seeders),
|
|
&pb.ReleaseInfo{
|
|
InfoHash: best.rel.InfoHash,
|
|
Format: best.rel.Format.String(),
|
|
BitDepth: int32(best.rel.BitDepth),
|
|
SampleRate: int32(best.rel.SampleRate),
|
|
Seeders: int32(best.item.Seeders),
|
|
Tracker: best.item.Tracker,
|
|
})
|
|
|
|
w.publisher.PublishStatus(ctx, pb.MonitorStep_MONITOR_STEP_ADDING_TORRENT,
|
|
fmt.Sprintf("Adding torrent: %s...", best.item.Title), nil)
|
|
|
|
if w.mode == pb.InteractionMode_INTERACTION_MODE_MANUAL {
|
|
decision, err := w.promptAndWait(ctx, &pb.PromptForDecision{
|
|
PromptId: w.nextPromptID(),
|
|
Type: pb.PromptType_PROMPT_TYPE_CONFIRM,
|
|
Message: fmt.Sprintf("Add torrent '%s' to client?", best.item.Title),
|
|
|
|
Options: &pb.PromptForDecision_Confirm{
|
|
Confirm: &pb.ConfirmPrompt{
|
|
ConfirmLabel: "Add",
|
|
CancelLabel: "Skip",
|
|
DefaultValue: true,
|
|
},
|
|
},
|
|
})
|
|
if err != nil {
|
|
w.publisher.PublishError(ctx, pb.MonitorStep_MONITOR_STEP_ADDING_TORRENT, err, false)
|
|
return err
|
|
}
|
|
if !decision.GetConfirm() {
|
|
w.publisher.PublishStatus(ctx, pb.MonitorStep_MONITOR_STEP_COMPLETE, "Skipped by user", nil)
|
|
return w.publisher.PublishResult(ctx, w.service.buildMonitorAlbumResponse(ctx, album, dbAlbum, nil))
|
|
}
|
|
}
|
|
|
|
w.publisher.PublishStatus(ctx, pb.MonitorStep_MONITOR_STEP_SAVING, "Saving to database...", nil)
|
|
|
|
dbAlbum, _ = w.service.metadata.GetAlbumByExternalID(ctx, album.GetId())
|
|
if dbAlbum != nil {
|
|
w.publisher.SetAlbumID(dbAlbum.ID)
|
|
w.service.saveTorrentAndDownload(ctx, dbAlbum.ID, best)
|
|
} else {
|
|
log.Warn().Str("album_id", w.req.AlbumId).Msg("album not in DB, skipping torrent/download persistence")
|
|
}
|
|
|
|
w.addedHash = best.rel.InfoHash
|
|
|
|
if err := w.service.addToTorrentClient(best); err != nil {
|
|
w.publisher.PublishError(ctx, pb.MonitorStep_MONITOR_STEP_ADDING_TORRENT, err, true)
|
|
return err
|
|
}
|
|
|
|
w.publisher.PublishStatus(ctx, pb.MonitorStep_MONITOR_STEP_COMPLETE, "Done!", nil)
|
|
|
|
return w.publisher.PublishResult(ctx, w.service.buildMonitorAlbumResponse(ctx, album, dbAlbum, &best))
|
|
}
|
|
|
|
func (w *monitorWorkflow) cleanup(ctx context.Context) {
|
|
if w.addedHash != "" {
|
|
if err := w.service.torrentClient.DeleteTorrent(w.addedHash); err != nil {
|
|
log.Warn().Err(err).Str("hash", w.addedHash).Msg("failed to delete torrent during cancel cleanup")
|
|
}
|
|
if err := w.service.downloads.SetCancelledByQbitHash(ctx, w.addedHash); err != nil {
|
|
log.Warn().Err(err).Str("hash", w.addedHash).Msg("failed to cancel download during cleanup")
|
|
}
|
|
}
|
|
}
|