0afef05099
Add a new archiver to download videos using yt-dlp. The new extractor should call the yt-dlp binary to download the video and we should track progress in the output and return code. The default rules should be updated so youtube videos are extracted using this extractor. The extractor default config should get the thumbnail as well (so we don't depend on the thumbnail extractor) and subtitles. Update the dockerfile accordingly so we not only have yt-dlp but it's required dependencies as well. Prefer installing from packages, if possible.
411 lines
14 KiB
Go
411 lines
14 KiB
Go
package domain
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"time"
|
|
|
|
"git.nakama.town/fmartingr/hako/internal/archival/archiver"
|
|
archivalRules "git.nakama.town/fmartingr/hako/internal/archival/rules"
|
|
archivalStore "git.nakama.town/fmartingr/hako/internal/archival/store"
|
|
"git.nakama.town/fmartingr/hako/internal/dependencies"
|
|
"git.nakama.town/fmartingr/hako/internal/model"
|
|
"github.com/google/uuid"
|
|
)
|
|
|
|
// ArchiveDomain handles business logic for archives
|
|
type ArchiveDomain struct {
|
|
deps model.Dependencies
|
|
}
|
|
|
|
// NewArchiveDomain creates a new ArchiveDomain
|
|
func NewArchiveDomain(deps model.Dependencies) model.ArchiveDomain {
|
|
return &ArchiveDomain{deps: deps}
|
|
}
|
|
|
|
// getDeps returns the concrete dependencies implementation
|
|
func (d *ArchiveDomain) getDeps() *dependencies.Dependencies {
|
|
return d.deps.(*dependencies.Dependencies)
|
|
}
|
|
|
|
// ConfigToMapConverter defines an interface for config types that can convert themselves to maps
|
|
type ConfigToMapConverter interface {
|
|
ToMap() map[string]any
|
|
}
|
|
|
|
// configToMap converts an archiver default config (struct or map) to map[string]any for ApplyConfig
|
|
func configToMap(cfg any) map[string]any {
|
|
// If it's already a map, return it directly
|
|
if m, ok := cfg.(map[string]any); ok {
|
|
return m
|
|
}
|
|
// If it implements ConfigToMapConverter, use its ToMap method
|
|
if converter, ok := cfg.(ConfigToMapConverter); ok {
|
|
return converter.ToMap()
|
|
}
|
|
// Fallback: use JSON marshaling/unmarshaling
|
|
data, err := json.Marshal(cfg)
|
|
if err != nil {
|
|
return nil
|
|
}
|
|
var out map[string]any
|
|
_ = json.Unmarshal(data, &out)
|
|
return out
|
|
}
|
|
|
|
// GetArchive retrieves a single archive by ID
|
|
func (d *ArchiveDomain) GetArchive(ctx context.Context, archiveID string) (*model.Archive, error) {
|
|
deps := d.getDeps()
|
|
archive, err := deps.ArchiveStore.GetByID(ctx, archiveID)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("failed to get archive: %w", err)
|
|
}
|
|
return archive, nil
|
|
}
|
|
|
|
// GetArchiveHistory retrieves all archives for a link
|
|
func (d *ArchiveDomain) GetArchiveHistory(ctx context.Context, linkID string) ([]*model.Archive, error) {
|
|
deps := d.getDeps()
|
|
opts := archivalStore.ArchiveListOptions{LinkID: linkID}
|
|
archives, err := deps.ArchiveStore.ListByLinkID(ctx, opts)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("failed to get archive history: %w", err)
|
|
}
|
|
return archives, nil
|
|
}
|
|
|
|
// ReArchiveLink creates a new archive for a link and enqueues the job
|
|
func (d *ArchiveDomain) ReArchiveLink(ctx context.Context, linkID string) error {
|
|
deps := d.getDeps()
|
|
// Get the link first to access userID and URL
|
|
storeLink, err := deps.LinkStore.GetByID(ctx, linkID)
|
|
if err != nil {
|
|
return fmt.Errorf("failed to get link: %w", err)
|
|
}
|
|
|
|
// Analyze URL using HEAD request to get current metadata
|
|
metadata, err := archivalRules.AnalyzeURL(ctx, storeLink.URL)
|
|
if err != nil {
|
|
return fmt.Errorf("failed to analyze URL: %w", err)
|
|
}
|
|
|
|
// Evaluate rules to determine extractors
|
|
extractors, err := deps.RuleEngine.Evaluate(ctx, metadata)
|
|
if err != nil {
|
|
return fmt.Errorf("failed to evaluate rules: %w", err)
|
|
}
|
|
|
|
// Log which extractors were selected
|
|
deps.Logger().Info("Rule engine selected extractors for re-archive", "url", storeLink.URL, "extractorCount", len(extractors), "extractors", extractors)
|
|
|
|
// If no extractors returned (empty array), skip archiving
|
|
if len(extractors) == 0 {
|
|
deps.Logger().Info("No extractors selected for re-archive, skipping", "url", storeLink.URL)
|
|
return fmt.Errorf("no extractors selected by rules")
|
|
}
|
|
|
|
now := time.Now()
|
|
// Create a new archive for re-archival
|
|
archive := &model.Archive{
|
|
ID: uuid.New().String(),
|
|
LinkID: linkID,
|
|
UserID: storeLink.UserID,
|
|
Status: model.ArchiveStatusPending,
|
|
CreatedAt: now,
|
|
}
|
|
|
|
if err := deps.ArchiveStore.Create(ctx, archive); err != nil {
|
|
return fmt.Errorf("failed to create archive: %w", err)
|
|
}
|
|
|
|
// Enqueue archive job with extractor keys
|
|
extractorKeys := make([]string, len(extractors))
|
|
for i, ext := range extractors {
|
|
extractorKeys[i] = ext.Key
|
|
}
|
|
|
|
payload := model.ArchiveLinkPayload{
|
|
ArchiveID: archive.ID,
|
|
LinkID: linkID,
|
|
ArchiverKeys: extractorKeys,
|
|
}
|
|
|
|
if err := deps.Queue.Enqueue(ctx, model.JobTypeArchiveLink, payload); err != nil {
|
|
return fmt.Errorf("failed to enqueue archive job: %w", err)
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// GetArchiveFiles retrieves all files for an archive
|
|
func (d *ArchiveDomain) GetArchiveFiles(ctx context.Context, archiveID string) ([]*model.ArchiveFile, error) {
|
|
deps := d.getDeps()
|
|
opts := archivalStore.ArchiveFileListOptions{ArchiveID: archiveID}
|
|
files, err := deps.ArchiveFileStore.ListByArchiveID(ctx, opts)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("failed to get archive files: %w", err)
|
|
}
|
|
return files, nil
|
|
}
|
|
|
|
// GetArchiveFile retrieves a specific archive file
|
|
func (d *ArchiveDomain) GetArchiveFile(ctx context.Context, fileID string) (*model.ArchiveFile, error) {
|
|
deps := d.getDeps()
|
|
file, err := deps.ArchiveFileStore.GetByID(ctx, fileID)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("failed to get archive file: %w", err)
|
|
}
|
|
return file, nil
|
|
}
|
|
|
|
// DeleteArchive deletes an archive and all its associated files
|
|
func (d *ArchiveDomain) DeleteArchive(ctx context.Context, archiveID string) error {
|
|
deps := d.getDeps()
|
|
// Get the archive first to verify it exists and get linkID
|
|
archive, err := deps.ArchiveStore.GetByID(ctx, archiveID)
|
|
if err != nil {
|
|
return fmt.Errorf("failed to get archive: %w", err)
|
|
}
|
|
|
|
// Get all archive files for this archive
|
|
opts := archivalStore.ArchiveFileListOptions{ArchiveID: archiveID}
|
|
storeFiles, err := deps.ArchiveFileStore.ListByArchiveID(ctx, opts)
|
|
if err != nil {
|
|
return fmt.Errorf("failed to list archive files: %w", err)
|
|
}
|
|
|
|
// Delete all storage files
|
|
for _, file := range storeFiles {
|
|
if err := deps.Storage.Delete(ctx, file.StoragePath); err != nil {
|
|
// Log error but continue - file might already be deleted
|
|
deps.Logger().Error("Failed to delete storage file", "path", file.StoragePath, "error", err)
|
|
}
|
|
}
|
|
|
|
// Delete the archive (cascade will delete archive files from database)
|
|
if err := deps.ArchiveStore.Delete(ctx, archiveID); err != nil {
|
|
return fmt.Errorf("failed to delete archive: %w", err)
|
|
}
|
|
|
|
// Recalculate total size for the link
|
|
if err := deps.LinkStore.RecalculateTotalSize(ctx, archive.LinkID, deps.ArchiveFileStore); err != nil {
|
|
// Log error but don't fail - total size update is not critical
|
|
deps.Logger().Error("Failed to recalculate total size for link after archive deletion", "linkID", archive.LinkID, "error", err)
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// ProcessArchive performs the actual archival operation (called by worker)
|
|
// If archiverKeys is provided, only those archivers will be used. Otherwise, all enabled archivers are used.
|
|
func (d *ArchiveDomain) ProcessArchive(ctx context.Context, archive *model.Archive, link *model.Link, archiverKeys []string) error {
|
|
deps := d.getDeps()
|
|
|
|
// Update status to processing
|
|
if err := deps.ArchiveStore.UpdateStatus(ctx, archive.ID, model.ArchiveStatusProcessing, ""); err != nil {
|
|
return fmt.Errorf("failed to update archive status: %w", err)
|
|
}
|
|
|
|
// Get archivers to use
|
|
var archiversToUse []archiver.Archiver
|
|
if len(archiverKeys) > 0 {
|
|
// Use specified archivers
|
|
archiversToUse = make([]archiver.Archiver, 0, len(archiverKeys))
|
|
for _, key := range archiverKeys {
|
|
arch, ok := deps.ArchiverMgr.GetArchiver(key)
|
|
if !ok {
|
|
deps.Logger().Warn("Archiver not found, skipping", "archiverKey", key)
|
|
continue
|
|
}
|
|
if !arch.IsEnabled(ctx) {
|
|
deps.Logger().Warn("Archiver is disabled, skipping", "archiverKey", key)
|
|
continue
|
|
}
|
|
archiversToUse = append(archiversToUse, arch)
|
|
}
|
|
if len(archiversToUse) == 0 {
|
|
err := fmt.Errorf("no valid archivers found from provided keys")
|
|
if updateErr := deps.ArchiveStore.UpdateStatus(ctx, archive.ID, model.ArchiveStatusFailed, err.Error()); updateErr != nil {
|
|
deps.Logger().Error("Failed to update archive status after no archivers found", "archiveID", archive.ID, "error", updateErr)
|
|
}
|
|
return err
|
|
}
|
|
} else {
|
|
// Use all enabled archivers (backward compatibility)
|
|
archiversToUse = deps.ArchiverMgr.ListEnabled(ctx)
|
|
if len(archiversToUse) == 0 {
|
|
err := fmt.Errorf("no enabled archivers available")
|
|
if updateErr := deps.ArchiveStore.UpdateStatus(ctx, archive.ID, model.ArchiveStatusFailed, err.Error()); updateErr != nil {
|
|
deps.Logger().Error("Failed to update archive status after no archivers found", "archiveID", archive.ID, "error", updateErr)
|
|
}
|
|
return err
|
|
}
|
|
}
|
|
|
|
// Track results per archiver
|
|
type archiverResult struct {
|
|
archiverKey string
|
|
success bool
|
|
title string
|
|
files int
|
|
error string
|
|
}
|
|
|
|
results := make([]archiverResult, 0, len(archiversToUse))
|
|
now := time.Now()
|
|
mimeTypes := make([]string, 0)
|
|
var firstTitle string
|
|
|
|
// Process each archiver
|
|
for _, arch := range archiversToUse {
|
|
archiverKey := arch.Key()
|
|
result := archiverResult{
|
|
archiverKey: archiverKey,
|
|
success: false,
|
|
}
|
|
|
|
// Load and apply archiver configuration
|
|
var config map[string]any
|
|
storedConfig, err := deps.ArchiverConfigStore.Get(ctx, archiverKey)
|
|
if err != nil {
|
|
deps.Logger().Warn("Failed to load archiver config, using default", "archiverKey", archiverKey, "error", err)
|
|
config = configToMap(arch.GetDefaultConfig())
|
|
} else if storedConfig != nil && storedConfig.ConfigJSON != "" {
|
|
// Parse stored config JSON
|
|
if err := json.Unmarshal([]byte(storedConfig.ConfigJSON), &config); err != nil {
|
|
deps.Logger().Warn("Failed to parse stored archiver config, using default", "archiverKey", archiverKey, "error", err)
|
|
config = configToMap(arch.GetDefaultConfig())
|
|
}
|
|
} else {
|
|
// No stored config, use default
|
|
config = configToMap(arch.GetDefaultConfig())
|
|
}
|
|
|
|
// Apply config to archiver
|
|
if err := arch.ApplyConfig(config); err != nil {
|
|
deps.Logger().Warn("Failed to apply archiver config, continuing with current settings", "archiverKey", archiverKey, "error", err)
|
|
// Continue anyway - archiver may have default settings
|
|
}
|
|
|
|
// Perform the archival
|
|
archiveResult, err := arch.Archive(ctx, link, deps.Storage)
|
|
if err != nil {
|
|
result.error = err.Error()
|
|
deps.Logger().Error("Archiver failed", "archiverKey", archiverKey, "archiveID", archive.ID, "error", err)
|
|
results = append(results, result)
|
|
continue
|
|
}
|
|
|
|
// Archiver succeeded
|
|
result.success = true
|
|
result.title = archiveResult.Title
|
|
result.files = len(archiveResult.Files)
|
|
|
|
// Save archive files with archiver key
|
|
for _, fileInfo := range archiveResult.Files {
|
|
archiveFile := &model.ArchiveFile{
|
|
ID: uuid.New().String(),
|
|
ArchiveID: archive.ID,
|
|
ArchiverKey: archiverKey,
|
|
Filename: fileInfo.Filename,
|
|
MimeType: fileInfo.MimeType,
|
|
FileSize: fileInfo.FileSize,
|
|
StoragePath: fileInfo.Path,
|
|
HashSha256: fileInfo.HashSha256,
|
|
CreatedAt: now,
|
|
}
|
|
|
|
if err := deps.ArchiveFileStore.Create(ctx, archiveFile); err != nil {
|
|
deps.Logger().Error("Failed to create archive file", "archiveID", archive.ID, "archiverKey", archiverKey, "filename", fileInfo.Filename, "error", err)
|
|
result.success = false
|
|
result.error = fmt.Sprintf("failed to save file metadata: %v", err)
|
|
break // Stop processing files for this archiver
|
|
}
|
|
|
|
// Enqueue content extraction job if MIME type is present
|
|
if fileInfo.MimeType != "" {
|
|
// Enqueue job to extract content (non-blocking, will be processed asynchronously)
|
|
extractPayload := map[string]string{
|
|
"file_id": archiveFile.ID,
|
|
}
|
|
if err := deps.Queue.Enqueue(ctx, model.JobTypeExtractContent, extractPayload); err != nil {
|
|
// Log error but don't fail the archive - content extraction is not critical
|
|
deps.Logger().Error("Failed to enqueue content extraction job", "fileID", archiveFile.ID, "error", err)
|
|
}
|
|
}
|
|
|
|
// Collect MIME types for category determination
|
|
if fileInfo.MimeType != "" {
|
|
mimeTypes = append(mimeTypes, fileInfo.MimeType)
|
|
}
|
|
}
|
|
|
|
// Store title from first successful archiver
|
|
if result.success && firstTitle == "" && result.title != "" {
|
|
firstTitle = result.title
|
|
}
|
|
|
|
results = append(results, result)
|
|
}
|
|
|
|
// Determine final archive status based on results
|
|
var successCount, failureCount int
|
|
var errorMessages []string
|
|
|
|
for _, result := range results {
|
|
if result.success {
|
|
successCount++
|
|
} else {
|
|
failureCount++
|
|
if result.error != "" {
|
|
errorMessages = append(errorMessages, fmt.Sprintf("%s: %s", result.archiverKey, result.error))
|
|
}
|
|
}
|
|
}
|
|
|
|
var finalStatus model.ArchiveStatus
|
|
var errorMessage string
|
|
|
|
if successCount == 0 {
|
|
// All archivers failed
|
|
finalStatus = model.ArchiveStatusFailed
|
|
errorMessage = fmt.Sprintf("All archivers failed: %s", fmt.Sprintf("%v", errorMessages))
|
|
} else if failureCount == 0 {
|
|
// All archivers succeeded
|
|
finalStatus = model.ArchiveStatusCompleted
|
|
} else {
|
|
// Some succeeded, some failed - partial status
|
|
finalStatus = model.ArchiveStatusPartial
|
|
errorMessage = fmt.Sprintf("Some archivers failed: %s", fmt.Sprintf("%v", errorMessages))
|
|
}
|
|
|
|
// Update archive title from first successful archiver
|
|
if firstTitle != "" {
|
|
if err := deps.ArchiveStore.UpdateTitle(ctx, archive.ID, firstTitle); err != nil {
|
|
deps.Logger().Error("Failed to update archive title", "archiveID", archive.ID, "error", err)
|
|
// Don't fail the archive for title update failure
|
|
}
|
|
}
|
|
|
|
// Update link categories based on file MIME types
|
|
if deps.Domains().Categories() != nil && len(mimeTypes) > 0 {
|
|
if err := deps.Domains().Categories().UpdateLinkCategories(ctx, link.ID, mimeTypes); err != nil {
|
|
// Log error but don't fail the archive - categories are not critical
|
|
deps.Logger().Error("Failed to update link categories", "linkID", link.ID, "error", err)
|
|
}
|
|
}
|
|
|
|
// Calculate and update total size for the link
|
|
if err := deps.LinkStore.RecalculateTotalSize(ctx, link.ID, deps.ArchiveFileStore); err != nil {
|
|
// Log error but don't fail the archive - total size update is not critical
|
|
deps.Logger().Error("Failed to recalculate total size for link", "linkID", link.ID, "error", err)
|
|
}
|
|
|
|
// Update archive status
|
|
if err := deps.ArchiveStore.UpdateStatus(ctx, archive.ID, finalStatus, errorMessage); err != nil {
|
|
return fmt.Errorf("failed to update archive status: %w", err)
|
|
}
|
|
|
|
return nil
|
|
}
|