Merge pull request #15 from teal-fm/natb/refactor-submissions

Refactor submissions into an atproto package

cmd/handlers.go

···
 
 	"github.com/teal-fm/piper/db"
 	"github.com/teal-fm/piper/models"
+	atprotoauth "github.com/teal-fm/piper/oauth/atproto"
 	pages "github.com/teal-fm/piper/pages"
+	atprotoservice "github.com/teal-fm/piper/service/atproto"
 	"github.com/teal-fm/piper/service/musicbrainz"
+	"github.com/teal-fm/piper/service/playingnow"
 	"github.com/teal-fm/piper/service/spotify"
 	"github.com/teal-fm/piper/session"
 )
···
 
 func apiMusicBrainzSearch(mbService *musicbrainz.MusicBrainzService) http.HandlerFunc {
 	return func(w http.ResponseWriter, r *http.Request) {
+		if mbService == nil {
+			jsonResponse(w, http.StatusServiceUnavailable, map[string]string{"error": "MusicBrainz service is not available"})
+			return
+		}
 
 		params := musicbrainz.SearchParams{
 			Track:   r.URL.Query().Get("track"),
···
 }
 
 // apiSubmitListensHandler handles ListenBrainz-compatible submissions
-func apiSubmitListensHandler(database *db.DB) http.HandlerFunc {
+func apiSubmitListensHandler(database *db.DB, atprotoService *atprotoauth.ATprotoAuthService, playingNowService *playingnow.PlayingNowService, mbService *musicbrainz.MusicBrainzService) http.HandlerFunc {
 	return func(w http.ResponseWriter, r *http.Request) {
 		userID, authenticated := session.GetUserID(r.Context())
 		if !authenticated {
···
 			return
 		}
 
+		// Get user for PDS submission
+		user, err := database.GetUserByID(userID)
+		if err != nil {
+			log.Printf("apiSubmitListensHandler: Error getting user %d: %v", userID, err)
+			jsonResponse(w, http.StatusInternalServerError, map[string]string{"error": "Failed to get user"})
+			return
+		}
+
 		// Process each listen in the payload
 		var processedTracks []models.Track
 		var errors []string
···
 			// Convert to internal Track format
 			track := listen.ConvertToTrack(userID)
 
-			// For 'playing_now' type, we might want to handle differently
-			// For now, treat all the same but could add temporary storage later
+			// Attempt to hydrate with MusicBrainz data if service is available and track doesn't have MBIDs
+			if mbService != nil && track.RecordingMBID == nil {
+				hydratedTrack, err := musicbrainz.HydrateTrack(mbService, track)
+				if err != nil {
+					log.Printf("apiSubmitListensHandler: Could not hydrate track with MusicBrainz for user %d: %v (continuing with original data)", userID, err)
+					// Continue with non-hydrated track
+				} else if hydratedTrack != nil {
+					track = *hydratedTrack
+					log.Printf("apiSubmitListensHandler: Successfully hydrated track '%s' with MusicBrainz data", track.Name)
+				}
+			}
+
+			// For 'playing_now' type, publish to PDS as actor status
 			if submission.ListenType == "playing_now" {
 				log.Printf("Received playing_now listen for user %d: %s - %s", userID, track.Artist[0].Name, track.Name)
-				// Could store in a separate playing_now table or just log
+
+				if user.ATProtoDID != nil && playingNowService != nil {
+					if err := playingNowService.PublishPlayingNow(r.Context(), userID, &track); err != nil {
+						log.Printf("apiSubmitListensHandler: Error publishing playing_now to PDS for user %d: %v", userID, err)
+						// Don't fail the request, just log the error
+					}
+				}
 				continue
 			}
 
···
 				log.Printf("apiSubmitListensHandler: Error saving track for user %d: %v", userID, err)
 				errors = append(errors, fmt.Sprintf("payload[%d]: failed to save track", i))
 				continue
+			}
+
+			// Submit to PDS as feed.play record
+			if user.ATProtoDID != nil && atprotoService != nil {
+				if err := atprotoservice.SubmitPlayToPDS(r.Context(), *user.ATProtoDID, &track, atprotoService); err != nil {
+					log.Printf("apiSubmitListensHandler: Error submitting play to PDS for user %d: %v", userID, err)
+					// Don't fail the request, just log the error
+				}
 			}
 
 			processedTracks = append(processedTracks, track)

cmd/listenbrainz_test.go

···
 
 	"github.com/teal-fm/piper/db"
 	"github.com/teal-fm/piper/models"
+	"github.com/teal-fm/piper/service/musicbrainz"
 	"github.com/teal-fm/piper/session"
 )
 
···
 	rr := httptest.NewRecorder()
 
 	// Call handler
-	handler := apiSubmitListensHandler(database)
+	handler := apiSubmitListensHandler(database, nil, nil, nil)
 	handler(rr, req)
 
 	// Check response
···
 	req = req.WithContext(ctx)
 
 	rr := httptest.NewRecorder()
-	handler := apiSubmitListensHandler(database)
+	handler := apiSubmitListensHandler(database, nil, nil, nil)
 	handler(rr, req)
 
 	if rr.Code != http.StatusOK {
···
 	req = req.WithContext(ctx)
 
 	rr := httptest.NewRecorder()
-	handler := apiSubmitListensHandler(database)
+	handler := apiSubmitListensHandler(database, nil, nil, nil)
 	handler(rr, req)
 
 	if rr.Code != http.StatusOK {
···
 	req = req.WithContext(ctx)
 
 	rr := httptest.NewRecorder()
-	handler := apiSubmitListensHandler(database)
+	handler := apiSubmitListensHandler(database, nil, nil, nil)
 	handler(rr, req)
 
 	if rr.Code != http.StatusOK {
···
 			req = req.WithContext(ctx)
 
 			rr := httptest.NewRecorder()
-			handler := apiSubmitListensHandler(database)
+			handler := apiSubmitListensHandler(database, nil, nil, nil)
 			handler(rr, req)
 
 			if rr.Code != tc.expectedStatus {
···
 	// No Authorization header
 
 	rr := httptest.NewRecorder()
-	handler := apiSubmitListensHandler(database)
+	handler := apiSubmitListensHandler(database, nil, nil, nil)
 	handler(rr, req)
 
 	if rr.Code != http.StatusUnauthorized {
···
 		t.Errorf("Second artist MBID not set correctly")
 	}
 }
+
+func TestListenBrainzSubmission_WithMusicBrainzHydration(t *testing.T) {
+	database := setupTestDB(t)
+	defer database.Close()
+
+	userID, apiKey := createTestUser(t, database)
+
+	// Create a MusicBrainz service for hydration
+	mbService := musicbrainz.NewMusicBrainzService(database)
+
+	// Create minimal submission (artist and track name only)
+	submission := models.ListenBrainzSubmission{
+		ListenType: "single",
+		Payload: []models.ListenBrainzPayload{
+			{
+				ListenedAt: func() *int64 { i := int64(1704067200); return &i }(),
+				TrackMetadata: models.ListenBrainzTrackMetadata{
+					ArtistName: "Daft Punk",
+					TrackName:  "One More Time",
+					// No MBIDs provided - should be hydrated
+				},
+			},
+		},
+	}
+
+	jsonData, err := json.Marshal(submission)
+	if err != nil {
+		t.Fatalf("Failed to marshal submission: %v", err)
+	}
+
+	req := httptest.NewRequest(http.MethodPost, "/1/submit-listens", bytes.NewReader(jsonData))
+	req.Header.Set("Content-Type", "application/json")
+	req.Header.Set("Authorization", "Token "+apiKey)
+
+	ctx := withUserContext(req.Context(), userID)
+	req = req.WithContext(ctx)
+
+	rr := httptest.NewRecorder()
+
+	// Call handler with MusicBrainz service
+	handler := apiSubmitListensHandler(database, nil, nil, mbService)
+	handler(rr, req)
+
+	if rr.Code != http.StatusOK {
+		t.Errorf("Expected status %d, got %d. Body: %s", http.StatusOK, rr.Code, rr.Body.String())
+	}
+
+	// Verify track was saved
+	tracks, err := database.GetRecentTracks(userID, 10)
+	if err != nil {
+		t.Fatalf("Failed to get tracks from database: %v", err)
+	}
+
+	if len(tracks) != 1 {
+		t.Fatalf("Expected 1 track in database, got %d", len(tracks))
+	}
+
+	track := tracks[0]
+
+	// The track should have been hydrated with MusicBrainz data
+	// Note: This test requires network access to MusicBrainz API
+	// In a real test environment, you might want to mock the HTTP client
+	if track.RecordingMBID != nil {
+		t.Logf("Track was hydrated with recording MBID: %s", *track.RecordingMBID)
+	}
+
+	if track.ReleaseMBID != nil {
+		t.Logf("Track was hydrated with release MBID: %s", *track.ReleaseMBID)
+	}
+
+	// Even if hydration fails, the track should still be saved with original data
+	if track.Name != "One More Time" {
+		t.Errorf("Expected track name 'One More Time', got %s", track.Name)
+	}
+	if len(track.Artist) == 0 || track.Artist[0].Name != "Daft Punk" {
+		t.Errorf("Expected artist 'Daft Punk', got %v", track.Artist)
+	}
+}

cmd/routes.go

···
 	mux.HandleFunc("/api/v1/musicbrainz/search", apiMusicBrainzSearch(app.mbService))                                     // MusicBrainz (public?)
 
 	// ListenBrainz-compatible endpoint
-	mux.HandleFunc("/1/submit-listens", session.WithAPIAuth(apiSubmitListensHandler(app.database), app.sessionManager))
+	mux.HandleFunc("/1/submit-listens", session.WithAPIAuth(apiSubmitListensHandler(app.database, app.atprotoService, app.playingNowService, app.mbService), app.sessionManager))
 
 	serverUrlRoot := viper.GetString("server.root_url")
 	atpClientId := viper.GetString("atproto.client_id")

service/atproto/submission.go

···
+package atproto
+
+import (
+	"context"
+	"fmt"
+	"log"
+	"time"
+
+	"github.com/bluesky-social/indigo/api/atproto"
+	lexutil "github.com/bluesky-social/indigo/lex/util"
+	"github.com/bluesky-social/indigo/xrpc"
+	"github.com/spf13/viper"
+	"github.com/teal-fm/piper/api/teal"
+	"github.com/teal-fm/piper/db"
+	"github.com/teal-fm/piper/models"
+	atprotoauth "github.com/teal-fm/piper/oauth/atproto"
+)
+
+// SubmitPlayToPDS submits a track play to the ATProto PDS as a feed.play record
+func SubmitPlayToPDS(ctx context.Context, did string, track *models.Track, atprotoService *atprotoauth.ATprotoAuthService) error {
+	if did == "" {
+		return fmt.Errorf("DID cannot be empty")
+	}
+
+	// Get ATProto client
+	client, err := atprotoService.GetATProtoClient()
+	if err != nil || client == nil {
+		return fmt.Errorf("failed to get ATProto client: %w", err)
+	}
+
+	xrpcClient := atprotoService.GetXrpcClient()
+	if xrpcClient == nil {
+		return fmt.Errorf("xrpc client is not available")
+	}
+
+	// Get user session
+	sess, err := atprotoService.DB.GetAtprotoSession(did, ctx, *client)
+	if err != nil {
+		return fmt.Errorf("couldn't get Atproto session for DID %s: %w", did, err)
+	}
+
+	// Convert track to feed.play record
+	playRecord, err := TrackToPlayRecord(track)
+	if err != nil {
+		return fmt.Errorf("failed to convert track to play record: %w", err)
+	}
+
+	// Create the record
+	input := atproto.RepoCreateRecord_Input{
+		Collection: "fm.teal.alpha.feed.play",
+		Repo:       sess.DID,
+		Record:     &lexutil.LexiconTypeDecoder{Val: playRecord},
+	}
+
+	authArgs := db.AtpSessionToAuthArgs(sess)
+	var out atproto.RepoCreateRecord_Output
+	if err := xrpcClient.Do(ctx, authArgs, xrpc.Procedure, "application/json", "com.atproto.repo.createRecord", nil, input, &out); err != nil {
+		return fmt.Errorf("failed to create play record for DID %s: %w", did, err)
+	}
+
+	log.Printf("Successfully submitted play to PDS for DID %s: %s - %s", did, track.Artist[0].Name, track.Name)
+	return nil
+}
+
+// TrackToPlayRecord converts a models.Track to teal.AlphaFeedPlay
+func TrackToPlayRecord(track *models.Track) (*teal.AlphaFeedPlay, error) {
+	if track.Name == "" {
+		return nil, fmt.Errorf("track name cannot be empty")
+	}
+
+	// Convert artists
+	artists := make([]*teal.AlphaFeedDefs_Artist, 0, len(track.Artist))
+	for _, a := range track.Artist {
+		artist := &teal.AlphaFeedDefs_Artist{
+			ArtistName: a.Name,
+			ArtistMbId: a.MBID,
+		}
+		artists = append(artists, artist)
+	}
+
+	// Prepare optional fields
+	var durationPtr *int64
+	if track.DurationMs > 0 {
+		durationSeconds := track.DurationMs / 1000
+		durationPtr = &durationSeconds
+	}
+
+	var playedTimeStr *string
+	if !track.Timestamp.IsZero() {
+		timeStr := track.Timestamp.Format(time.RFC3339)
+		playedTimeStr = &timeStr
+	}
+
+	var isrcPtr *string
+	if track.ISRC != "" {
+		isrcPtr = &track.ISRC
+	}
+
+	var originUrlPtr *string
+	if track.URL != "" {
+		originUrlPtr = &track.URL
+	}
+
+	var servicePtr *string
+	if track.ServiceBaseUrl != "" {
+		servicePtr = &track.ServiceBaseUrl
+	}
+
+	var releaseNamePtr *string
+	if track.Album != "" {
+		releaseNamePtr = &track.Album
+	}
+
+	// Get submission client agent
+	submissionAgent := viper.GetString("app.submission_agent")
+	if submissionAgent == "" {
+		submissionAgent = "piper/v0.0.1"
+	}
+
+	playRecord := &teal.AlphaFeedPlay{
+		LexiconTypeID:          "fm.teal.alpha.feed.play",
+		TrackName:              track.Name,
+		Artists:                artists,
+		Duration:               durationPtr,
+		PlayedTime:             playedTimeStr,
+		RecordingMbId:          track.RecordingMBID,
+		ReleaseMbId:            track.ReleaseMBID,
+		ReleaseName:            releaseNamePtr,
+		Isrc:                   isrcPtr,
+		OriginUrl:              originUrlPtr,
+		MusicServiceBaseDomain: servicePtr,
+		SubmissionClientAgent:  &submissionAgent,
+	}
+
+	return playRecord, nil
+}

service/lastfm/lastfm.go

···
 import (
 	"context"
 	"encoding/json"
-	"errors"
 	"fmt"
 	"io"
 	"log"
···
 	"sync"
 	"time"
 
-	"github.com/bluesky-social/indigo/api/atproto"
-	lexutil "github.com/bluesky-social/indigo/lex/util"
-	"github.com/bluesky-social/indigo/xrpc"
-	"github.com/spf13/viper"
-	"github.com/teal-fm/piper/api/teal"
 	"github.com/teal-fm/piper/db"
 	"github.com/teal-fm/piper/models"
 	atprotoauth "github.com/teal-fm/piper/oauth/atproto"
+	atprotoservice "github.com/teal-fm/piper/service/atproto"
 	"github.com/teal-fm/piper/service/musicbrainz"
 	"golang.org/x/time/rate"
 )
···
 }
 
 func (l *LastFMService) SubmitTrackToPDS(did string, track *models.Track, ctx context.Context) error {
-	client, err := l.atprotoService.GetATProtoClient()
-	if err != nil || client == nil {
-		return err
-	}
-
-	xrpcClient := l.atprotoService.GetXrpcClient()
-	if xrpcClient == nil {
-		return errors.New("xrpc client is kil")
-	}
-
-	// we check for client above
-	sess, err := l.db.GetAtprotoSession(did, ctx, *client)
-	if err != nil {
-		return fmt.Errorf("Couldn't get Atproto session: %s", err)
-	}
-
-	// printout the session details
-	l.logger.Printf("Submitting track for the did: %+v\n", sess.DID)
-
-	artists := make([]*teal.AlphaFeedDefs_Artist, 0, len(track.Artist))
-	for _, a := range track.Artist {
-		artist := &teal.AlphaFeedDefs_Artist{
-			ArtistName: a.Name,
-			ArtistMbId: a.MBID,
-		}
-		artists = append(artists, artist)
-	}
-
-	var durationPtr *int64
-	if track.DurationMs > 0 {
-		durationSeconds := track.DurationMs / 1000
-		durationPtr = &durationSeconds
-	}
-
-	playedTimeStr := track.Timestamp.Format(time.RFC3339)
-	submissionAgent := viper.GetString("app.submission_agent")
-	if submissionAgent == "" {
-		submissionAgent = "piper/v0.0.1" // Default if not configured
-	}
-
-	// track -> tealfm track
-	tfmTrack := teal.AlphaFeedPlay{
-		LexiconTypeID: "fm.teal.alpha.feed.play", // Assuming this is the correct Lexicon ID
-		// tfm specifies duration in seconds
-		Duration:  durationPtr, // Pointer required
-		TrackName: track.Name,
-		// should be unix timestamp
-		PlayedTime:            &playedTimeStr, // Pointer required
-		Artists:               artists,
-		ReleaseMbId:           track.ReleaseMBID,   // Pointer required
-		ReleaseName:           &track.Album,        // Pointer required
-		RecordingMbId:         track.RecordingMBID, // Pointer required
-		SubmissionClientAgent: &submissionAgent,    // Pointer required
-	}
-
-	input := atproto.RepoCreateRecord_Input{
-		Collection: "fm.teal.alpha.feed.play",
-		Repo:       sess.DID,
-		Record:     &lexutil.LexiconTypeDecoder{Val: &tfmTrack},
-	}
-
-	authArgs := db.AtpSessionToAuthArgs(sess)
-
-	var out atproto.RepoCreateRecord_Output
-	if err := xrpcClient.Do(ctx, authArgs, xrpc.Procedure, "application/json", "com.atproto.repo.createRecord", nil, input, &out); err != nil {
-		return err
-	}
-
-	// submit track to PDS
-
-	return nil
+	// Use shared atproto service for submission
+	return atprotoservice.SubmitPlayToPDS(ctx, did, track, l.atprotoService)
 }
 
 // convertLastFMTrackToModelsTrack converts a Last.fm Track to models.Track format

service/spotify/spotify.go

···
 
 	"context" // Added for context.Context
 
-	"github.com/bluesky-social/indigo/api/atproto"      // Added for atproto.RepoCreateRecord_Input
-	lexutil "github.com/bluesky-social/indigo/lex/util" // Added for lexutil.LexiconTypeDecoder
-	"github.com/bluesky-social/indigo/xrpc"             // Added for xrpc.Client
-	"github.com/spf13/viper"
-	"github.com/teal-fm/piper/api/teal" // Added for teal.AlphaFeedPlay
+	// Added for atproto.RepoCreateRecord_Input
+	// Added for lexutil.LexiconTypeDecoder
+	// Added for xrpc.Client
+	"github.com/spf13/viper" // Added for teal.AlphaFeedPlay
 	"github.com/teal-fm/piper/db"
 	"github.com/teal-fm/piper/models"
 	atprotoauth "github.com/teal-fm/piper/oauth/atproto"
+	atprotoservice "github.com/teal-fm/piper/service/atproto"
 	"github.com/teal-fm/piper/service/musicbrainz"
 	"github.com/teal-fm/piper/session"
 )
···
 }
 
 func (s *SpotifyService) SubmitTrackToPDS(did string, track *models.Track, ctx context.Context) error {
-	client, err := s.atprotoService.GetATProtoClient()
-	if err != nil || client == nil {
-		s.logger.Printf("Error getting ATProto client: %v", err)
-		return fmt.Errorf("failed to get ATProto client: %w", err)
-	}
-
-	xrpcClient := s.atprotoService.GetXrpcClient()
-	if xrpcClient == nil {
-		return errors.New("xrpc client is not available")
-	}
-
-	sess, err := s.DB.GetAtprotoSession(did, ctx, *client)
-	if err != nil {
-		return fmt.Errorf("couldn't get Atproto session for DID %s: %w", did, err)
-	}
-
-	artists := make([]*teal.AlphaFeedDefs_Artist, 0, len(track.Artist))
-	for _, a := range track.Artist {
-		artist := &teal.AlphaFeedDefs_Artist{
-			ArtistName: a.Name,
-			ArtistMbId: a.MBID,
-		}
-		artists = append(artists, artist)
-	}
-
-	var durationPtr *int64
-	if track.DurationMs > 0 {
-		durationSeconds := track.DurationMs / 1000
-		durationPtr = &durationSeconds
-	}
-
-	playedTimeStr := track.Timestamp.Format(time.RFC3339)
-	submissionAgent := viper.GetString("app.submission_agent")
-	if submissionAgent == "" {
-		submissionAgent = "piper/v0.0.1" // Default if not configured
-	}
-
 	//Had a empty feed.play get submitted not sure why. Tracking here
 	if track.Name == "" {
 		s.logger.Println("Track name is empty. Skipping submission. Please record the logs before and send to the teal.fm Discord")
 		return nil
 	}
 
-	tfmTrack := teal.AlphaFeedPlay{
-		LexiconTypeID: "fm.teal.alpha.feed.play",
-		Duration:      durationPtr,
-		TrackName:     track.Name,
-		PlayedTime:    &playedTimeStr,
-		Artists:       artists,
-		ReleaseMbId:   track.ReleaseMBID,
-		ReleaseName:   &track.Album,
-		RecordingMbId: track.RecordingMBID,
-		// Optional: Spotify specific data if your lexicon supports it
-		// SpotifyTrackID: &track.ServiceID,
-		// SpotifyAlbumID: &track.ServiceAlbumID,
-		// SpotifyArtistIDs: track.ServiceArtistIDs, // Assuming this is a []string
-		SubmissionClientAgent: &submissionAgent,
-	}
-
-	input := atproto.RepoCreateRecord_Input{
-		Collection: "fm.teal.alpha.feed.play", // Ensure this collection is correct
-		Repo:       sess.DID,
-		Record:     &lexutil.LexiconTypeDecoder{Val: &tfmTrack},
-	}
-
-	authArgs := db.AtpSessionToAuthArgs(sess)
-
-	var out atproto.RepoCreateRecord_Output
-	if err := xrpcClient.Do(ctx, authArgs, xrpc.Procedure, "application/json", "com.atproto.repo.createRecord", nil, input, &out); err != nil {
-		s.logger.Printf("Error creating record for DID %s: %v. Input: %+v", did, err, input)
-		return fmt.Errorf("failed to create record on PDS for DID %s: %w", did, err)
-	}
-
-	s.logger.Printf("Successfully submitted track '%s' to PDS for DID %s. Record URI: %s", track.Name, did, out.Uri)
-	return nil
+	// Use shared atproto service for submission
+	return atprotoservice.SubmitPlayToPDS(ctx, did, track, s.atprotoService)
 }
 
 func (s *SpotifyService) SetAccessToken(token string, refreshToken string, userId int64, hasSession bool) (int64, error) {