Re: [PATCH] Add offline scrobble queueing to listenbrainz scrobbler

Fran, can you please have a look at this patch to your work?


coopi <[email protected]> writes:

>>From 3381b0af16d02814ce1e254dcf49553ad73ad2f1 Mon Sep 17 00:00:00 2001
> Message-ID: <[email protected]>
> MIME-Version: 1.0
> Content-Type: text/plain
>
>
> Queue failed listen submissions to a local file and flush them to ListenBrainz
> when connectivity is restored.  Failed HTTPS connections hang silently with no
> callback invocation; handle this by starting a per-request connection timeout
> that queues the listen and kills the hung connection if no response arrives.
>
> * emms-listenbrainz-scrobbler.el (emms-listenbrainz-scrobbler-queue-file): New
> defcustom.
> (emms-listenbrainz-scrobbler-flush-interval): New defcustom.
> (emms-listenbrainz-scrobbler-max-batch-size): New defcustom.
> (emms-listenbrainz-scrobbler-queue-timer): New variable.
> (emms-listenbrainz-scrobbler-connection-timeout): New variable.
> (emms-listenbrainz-scrobbler-flushing): New variable.
> (emms-listenbrainz-scrobbler--completed): New defvar-local.
> (emms-listenbrainz-scrobbler--queue-listen): New function.
> (emms-listenbrainz-scrobbler--read-queue): New function.
> (emms-listenbrainz-scrobbler--write-queue): New function.
> (emms-listenbrainz-scrobbler--make-import-payload): New function.
> (emms-listenbrainz-scrobbler--handle-submission-timeout): New function.
> (emms-listenbrainz-scrobbler--handle-flush-timeout): New function.
> (emms-listenbrainz-scrobbler--flush-callback): New function.
> (emms-listenbrainz-scrobbler--flush-batch): New function.
> (emms-listenbrainz-scrobbler-flush-queue): New function.
> (emms-listenbrainz-scrobbler-queue-count): New function.
> (emms-listenbrainz-scrobbler-clear-queue): New function.
> (emms-listenbrainz-scrobbler-make-query): Fix `duration' only being added to
> extra-data when `track-number' was set.
> (emms-listenbrainz-scrobbler-make-async-submission-call): Wrap `url-retrieve' in
> `condition-case' to catch synchronous network errors.  Extract listen data from
> the payload and pass it to the callback for potential offline queueing.  Start a
> connection timeout timer; on expiry, queue the listen and kill the hung
> connection.
> (emms-listenbrainz-scrobbler-submission-callback): Change argument list
> from (STATUS &optional CBARGS) to (STATUS &optional TRACK PLAYING-NOW
> LISTEN-DATA).  Check `emms-listenbrainz-scrobbler--completed' to avoid
> double-queueing when a timeout kills the connection before the callback fires.
> Check `plist-get STATUS :error' for network failures and queue failed listens
> for later submission.
> (emms-listenbrainz-scrobbler-enable): Start periodic queue flush timer.  Flush
> any queued listens from previous sessions.
> (emms-listenbrainz-scrobbler-disable): Cancel periodic flush timer.
> ---
>  emms-listenbrainz-scrobbler.el | 375 ++++++++++++++++++++++++++++++---
>  1 file changed, 344 insertions(+), 31 deletions(-)
>
> diff --git a/emms-listenbrainz-scrobbler.el b/emms-listenbrainz-scrobbler.el
> index 2262d4e..0d30f5f 100644
> --- a/emms-listenbrainz-scrobbler.el
> +++ b/emms-listenbrainz-scrobbler.el
> @@ -67,6 +67,47 @@ Note that the preferred way of authenticating is using authinfo.")
>    nil
>    "Non-nil if emms-listenbrainz-scrobbler is active.")
>  
> +(defcustom emms-listenbrainz-scrobbler-queue-file
> +  (expand-file-name "listenbrainz-queue" emms-directory)
> +  "File for storing offline listen queue."
> +  :type 'file
> +  :group 'emms)
> +
> +(defcustom emms-listenbrainz-scrobbler-flush-interval
> +  (* 10 60)                             ; 10 minutes
> +  "Interval in seconds between automatic queue flush attempts.
> +Set to nil to disable periodic flushing."
> +  :type '(choice (integer :tag "Seconds")
> +                 (const :tag "Disable periodic flush" nil))
> +  :group 'emms)
> +
> +(defcustom emms-listenbrainz-scrobbler-max-batch-size
> +  100
> +  "Maximum number of listens to submit in a single import request.
> +ListenBrainz accepts up to 1000, but smaller batches are gentler
> +on the server."
> +  :type 'integer
> +  :group 'emms)
> +
> +(defvar emms-listenbrainz-scrobbler-queue-timer
> +  nil
> +  "Timer for periodic queue flushing.")
> +
> +(defvar emms-listenbrainz-scrobbler-connection-timeout
> +  10
> +  "Seconds to wait for a connection before treating it as failed.")
> +
> +(defvar emms-listenbrainz-scrobbler-flushing
> +  nil
> +  "Non-nil if a queue flush is currently in progress.
> +Prevents overlapping flush operations.")
> +
> +(defvar-local emms-listenbrainz-scrobbler--completed
> +  nil
> +  "Non-nil when the HTTP callback has already handled this request.
> +Set to t by the callback, or 'timeout by the timeout handler.  Used to
> +prevent double-queueing when a timeout kills a hung connection.")
> +
>  
>  ;;* User token
>  
> @@ -110,7 +151,7 @@ Note that the preferred way of authenticating is using authinfo.")
>      ;; additional data
>      (when track-number (push (cons "tracknumber" (string-to-number track-number))
>  			     extra-data))
> -    (when track-number (push (cons "duration" track-length) extra-data))
> +    (when track-length (push (cons "duration" track-length) extra-data))
>      (when extra-data (push (cons "additional_info" extra-data) metadata))
>      ;; payload
>      (push (cons "track_metadata" metadata) payload)
> @@ -156,8 +197,12 @@ Note that the preferred way of authenticating is using authinfo.")
>  ;; data but headers!
>  
>  (defun emms-listenbrainz-scrobbler-make-async-submission-call (track &optional playing-now)
> -  "Submit listen, or playing-now if PLAYING-NOW non-nil, of TRACK to listenbrainz."
> +  "Submit listen, or playing-now if PLAYING-NOW non-nil, of TRACK to ListenBrainz.
> +On network failure, the listen is queued to the offline queue (unless
> +PLAYING-NOW is non-nil, since playing-now notifications are ephemeral)."
>    (let* ((payload (emms-listenbrainz-scrobbler-make-query track playing-now))
> +         ;; Extract the inner listen data (for offline queueing on failure)
> +	 (listen-data (aref (cdr (assoc "payload" payload)) 0))
>  	 (token (emms-listenbrainz-scrobbler-get-token))
>  	 (token-string (encode-coding-string (concat "Token " token) 'utf-8))
>  	 (url-request-method "POST")
> @@ -167,35 +212,287 @@ Note that the preferred way of authenticating is using authinfo.")
>      (ignore url-request-method
>  	    url-request-data
>  	    url-request-extra-headers)
> -    (url-retrieve
> -     emms-listenbrainz-scrobbler-submission-url
> -     #'emms-listenbrainz-scrobbler-submission-callback
> -     (list (cons track playing-now)))))
> +    (condition-case _
> +	(let ((buffer (url-retrieve
> +		       emms-listenbrainz-scrobbler-submission-url
> +		       #'emms-listenbrainz-scrobbler-submission-callback
> +		       (list track playing-now listen-data))))
> +	  (when (buffer-live-p buffer)
> +	    (with-current-buffer buffer
> +	      (setq emms-listenbrainz-scrobbler--completed nil))
> +	    (run-at-time emms-listenbrainz-scrobbler-connection-timeout nil
> +			 #'emms-listenbrainz-scrobbler--handle-submission-timeout
> +			 buffer track playing-now listen-data)))
> +      (error
> +       ;; Synchronous error (e.g. DNS resolution failure on some platforms).
> +       ;; `url-retrieve' signaled an error before the callback was invoked.
> +       (unless playing-now
> +	 (emms-listenbrainz-scrobbler--queue-listen listen-data)
> +	 (message "Listenbrainz: connection error, queued for later: %s"
> +		  (emms-track-get track 'info-title)))))))
> +
> +(defun emms-listenbrainz-scrobbler--handle-submission-timeout
> +    (buffer track playing-now listen-data)
> +  "Handle a submission connection timeout for BUFFER.
> +If the callback has not yet handled this request, queue the listen and
> +kill the hung connection."
> +  (when (and (buffer-live-p buffer)
> +	     (not (buffer-local-value
> +		   'emms-listenbrainz-scrobbler--completed buffer)))
> +    (with-current-buffer buffer
> +      (setq emms-listenbrainz-scrobbler--completed 'timeout))
> +    (kill-buffer buffer)
> +    (unless playing-now
> +      (emms-listenbrainz-scrobbler--queue-listen listen-data)
> +      (message "Listenbrainz: connection timed out, queued: %s"
> +	       (emms-track-get track 'info-title)))))
>  
> -(defun emms-listenbrainz-scrobbler-submission-callback (status &optional cbargs)
> +(defun emms-listenbrainz-scrobbler-submission-callback
> +    (status &optional track playing-now listen-data)
>    "Callback to handle response from listenbrainz server.
> +STATUS is the `url-retrieve' status plist.  TRACK is the EMMS track.
> +PLAYING-NOW is non-nil if this was a playing_now submission.
> +LISTEN-DATA is the serialized listen object for offline queueing."
> +  (if (eq emms-listenbrainz-scrobbler--completed 'timeout)
> +      ;; Already handled by the timeout handler; buffer is being killed.
> +      nil
> +    (setq emms-listenbrainz-scrobbler--completed t)
> +    (if (plist-get status :error)
> +	;; Network error: queue for later and bail out
> +	(unless playing-now
> +	  (emms-listenbrainz-scrobbler--queue-listen listen-data)
> +	  (message "Listenbrainz: network error, queued for later: %s"
> +		   (emms-track-get track 'info-title)))
> +      (when (< (point-max) 1)
> +	(error "No response from submission server"))
> +      (goto-char (point-min))
> +      (let* ((response (ignore-errors
> +			 (re-search-forward "\n\n")
> +			 (json-read)))
> +	     (title (emms-track-get track 'info-title)))
> +	(cond ((string= "ok" (alist-get 'status response))
> +	       (when (and (not playing-now)
> +			  emms-listenbrainz-scrobbler-display-submissions)
> +		 (message "Listenbrainz: submitted %s." title))
> +	       (kill-buffer))
> +	      ((assoc 'error response)
> +	       (unless playing-now
> +		 (emms-listenbrainz-scrobbler--queue-listen listen-data))
> +	       (message "Listenbrainz error: %s while submitting %s"
> +			(alist-get 'error response)
> +			title))
> +	      (t
> +	       (unless playing-now
> +		 (emms-listenbrainz-scrobbler--queue-listen listen-data))
> +	       (error "Listenbrainz: unhandled error while submitting %s" title)))))))
> +
> +
> +;;* Offline Queue
> +
> +;;; Storage
> +
> +(defun emms-listenbrainz-scrobbler--queue-listen (listen-data)
> +  "Append LISTEN-DATA to the offline queue file.
> +LISTEN-DATA should be an alist suitable for ListenBrainz submission,
> +i.e. it must have \"listened_at\" and \"track_metadata\" keys."
> +  (when listen-data
> +    (with-temp-buffer
> +      (prin1 listen-data (current-buffer))
> +      (insert "\n")
> +      (condition-case err
> +	  (append-to-file (point-min) (point-max)
> +			  emms-listenbrainz-scrobbler-queue-file)
> +	(error
> +	 (message "Listenbrainz: could not write to queue file: %s"
> +		  (error-message-string err)))))))
> +
> +(defun emms-listenbrainz-scrobbler--read-queue ()
> +  "Read all listens from the queue file.
> +Return a list of listen alists.  Each alist has \"listened_at\" and
> +\"track_metadata\" keys, suitable for ListenBrainz import."
> +  (let ((file emms-listenbrainz-scrobbler-queue-file))
> +    (when (file-exists-p file)
> +      (with-temp-buffer
> +	(insert-file-contents file)
> +	(goto-char (point-min))
> +	(let (listens)
> +	  (while (not (eobp))
> +	    (let ((start-pos (point))
> +		  (line (buffer-substring-no-properties
> +			 (line-beginning-position) (line-end-position))))
> +	      (if (string-empty-p (string-trim line))
> +		  (forward-line)
> +		(condition-case err
> +		    (progn
> +		      (push (read line) listens)
> +		      (forward-line))
> +		  (error
> +		   (message "Listenbrainz: skipping malformed queue entry: %s"
> +			    (error-message-string err))
> +		   (forward-line))))
> +	      ;; Safety: ensure we always advance past the current line
> +	      (when (and (not (eobp))
> +			 (= (point) start-pos))
> +		(forward-line))))
> +	  (nreverse listens))))))
> +
> +(defun emms-listenbrainz-scrobbler--write-queue (listens)
> +  "Write LISTENS to the queue file, replacing any existing content.
> +If LISTENS is nil, the queue file is deleted."
> +  (if (null listens)
> +      (when (file-exists-p emms-listenbrainz-scrobbler-queue-file)
> +	(delete-file emms-listenbrainz-scrobbler-queue-file))
> +    (with-temp-buffer
> +      (dolist (listen listens)
> +        (prin1 listen (current-buffer))
> +	(insert "\n"))
> +      (write-region (point-min) (point-max)
> +		    emms-listenbrainz-scrobbler-queue-file nil 'silent))))
>  
> -Ignore STATUS argument and store submission data in CBARGS."
> -  (ignore status)
> -  (when (< (point-max) 1)
> -    (error "No response from submission server"))
> -  (goto-char (point-min ))
> -  (let* ((response (ignore-errors
> -		     (re-search-forward "\n\n")
> -		     (json-read)))
> -	 (track (car cbargs))
> -	 (playing-now (cdr cbargs))
> -	 (title (emms-track-get track 'info-title)))
> -    (cond ((string= "ok" (alist-get 'status response))
> -	   (when (and (not playing-now) emms-listenbrainz-scrobbler-display-submissions)
> -	     (message "Listenbrainz: submitted %s." title))
> -	   ;; tidy up
> -	   (kill-buffer))
> -	  ((assoc 'error response)
> -	   (message "Listenbrainz error: %s while submitting %s"
> -		    (alist-get 'error response)
> -		    title))
> -	  (t (error "Listenbrainz: unhandled error while submitting %s" title)))))
> +
> +;;; Flushing
> +
> +(defun emms-listenbrainz-scrobbler--make-import-payload (listens)
> +  "Create a ListenBrainz import payload from a list of LISTEN alists.
> +Returns an alist with \"listen_type\" set to \"import\" and a
> +\"payload\" vector containing the listens."
> +  `(("listen_type" . "import")
> +    ("payload" . ,(apply #'vector listens))))
> +
> +(defun emms-listenbrainz-scrobbler-flush-queue ()
> +  "Submit all queued (offline) listens to ListenBrainz.
> +Uses the \"import\" listen type to submit listens in batches.  This is
> +safe to call at any time; it does nothing if the queue is empty or if a
> +flush is already in progress."
> +  (interactive)
> +  (unless emms-listenbrainz-scrobbler-flushing
> +    (let ((listens (emms-listenbrainz-scrobbler--read-queue)))
> +      (when listens
> +	(setq emms-listenbrainz-scrobbler-flushing t)
> +	(message "Listenbrainz: flushing %d queued listens..."
> +		 (length listens))
> +	(emms-listenbrainz-scrobbler--flush-batch listens 0)))))
> +
> +(defun emms-listenbrainz-scrobbler--handle-flush-timeout
> +    (buffer remaining-listens)
> +  "Handle a flush connection timeout for BUFFER.
> +REMAINING-LISTENS is all unsubmitted listens (including the timed-out
> +batch).  Write them back to the queue and reset the flushing flag."
> +  (when (and (buffer-live-p buffer)
> +	     (not (buffer-local-value
> +		   'emms-listenbrainz-scrobbler--completed buffer)))
> +    (with-current-buffer buffer
> +      (setq emms-listenbrainz-scrobbler--completed 'timeout))
> +    (kill-buffer buffer)
> +    (message "Listenbrainz: connection timed out during flush, %d listens remain queued."
> +	     (length remaining-listens))
> +    (emms-listenbrainz-scrobbler--write-queue remaining-listens)
> +    (setq emms-listenbrainz-scrobbler-flushing nil)))
> +
> +(defun emms-listenbrainz-scrobbler--flush-callback (status batch rest submitted-count)
> +  "Handle flush batch response from ListenBrainz server.
> +STATUS is the `url-retrieve' status plist.  BATCH is the list of listens
> +that were just submitted.  REST is the remaining unsubmitted listens.
> +SUBMITTED-COUNT is the number of listens already submitted in this flush
> +session."
> +  (if (eq emms-listenbrainz-scrobbler--completed 'timeout)
> +      ;; Already handled by the timeout handler.
> +      nil
> +    (setq emms-listenbrainz-scrobbler--completed t)
> +    (let ((err (plist-get status :error)))
> +      (if err
> +	  ;; Network error: write everything remaining back to queue
> +	  (progn
> +	    (message
> +	     "Listenbrainz: network error during flush, %d listens remain queued."
> +	     (+ (length batch) (length rest)))
> +	    (emms-listenbrainz-scrobbler--write-queue
> +	     (append batch rest))
> +	    (setq emms-listenbrainz-scrobbler-flushing nil)
> +	    (condition-case nil (kill-buffer) (error nil)))
> +	;; Connection succeeded: check HTTP response
> +	(goto-char (point-min))
> +	(let ((response (ignore-errors
> +			  (re-search-forward "\n\n")
> +			  (json-read))))
> +	  (condition-case nil (kill-buffer) (error nil))
> +	  (if (string= "ok" (alist-get 'status response))
> +	      ;; Batch accepted: continue with rest
> +	      (emms-listenbrainz-scrobbler--flush-batch
> +	       rest
> +	       (+ submitted-count (length batch)))
> +	    ;; API error: write everything remaining back to queue
> +	    (message
> +	     "Listenbrainz: API error during flush, %d listens remain queued."
> +	     (+ (length batch) (length rest)))
> +	    (emms-listenbrainz-scrobbler--write-queue
> +	     (append batch rest))
> +	    (setq emms-listenbrainz-scrobbler-flushing nil)))))))
> +
> +(defun emms-listenbrainz-scrobbler--flush-batch (remaining-listens submitted-count)
> +  "Submit the next batch of REMAINING-LISTENS to ListenBrainz.
> +SUBMITTED-COUNT is the number of listens already submitted in this flush
> +session.  On success, continues with the rest of the queue.  On failure,
> +writes remaining listens back to the queue file."
> +  (if (null remaining-listens)
> +      ;; All done
> +      (progn
> +	(emms-listenbrainz-scrobbler--write-queue nil)
> +	(setq emms-listenbrainz-scrobbler-flushing nil)
> +	(when (> submitted-count 0)
> +	  (message "Listenbrainz: flushed %d queued listens." submitted-count)))
> +    ;; Submit next batch
> +    (let* ((batch-size (min (length remaining-listens)
> +			    emms-listenbrainz-scrobbler-max-batch-size))
> +	   (batch (seq-take remaining-listens batch-size))
> +	   (rest (seq-drop remaining-listens batch-size))
> +	   (payload (emms-listenbrainz-scrobbler--make-import-payload batch))
> +	   (token (emms-listenbrainz-scrobbler-get-token))
> +	   (token-string (encode-coding-string (concat "Token " token) 'utf-8))
> +	   (url-request-method "POST")
> +	   (url-request-data (encode-coding-string (json-encode payload) 'utf-8))
> +	   (url-request-extra-headers
> +	    `(("Content-type" . "application/json; charset=utf-8")
> +	      ("Authorization" . ,token-string))))
> +      (ignore url-request-method
> +	      url-request-data
> +	      url-request-extra-headers)
> +      (condition-case _
> +	  (let ((buffer (url-retrieve
> +			 emms-listenbrainz-scrobbler-submission-url
> +			 #'emms-listenbrainz-scrobbler--flush-callback
> +			 (list batch rest submitted-count))))
> +	    (when (buffer-live-p buffer)
> +	      (with-current-buffer buffer
> +		(setq-local emms-listenbrainz-scrobbler--completed nil))
> +	      (run-at-time emms-listenbrainz-scrobbler-connection-timeout nil
> +			   #'emms-listenbrainz-scrobbler--handle-flush-timeout
> +			   buffer remaining-listens)))
> +	(error
> +	 ;; Synchronous error (e.g. DNS failure)
> +	 (message "Listenbrainz: connection error during flush, %d listens remain queued."
> +		  (length remaining-listens))
> +	 (emms-listenbrainz-scrobbler--write-queue remaining-listens)
> +	 (setq emms-listenbrainz-scrobbler-flushing nil))))))
> +
> +
> +;;; Interactive Commands
> +
> +(defun emms-listenbrainz-scrobbler-queue-count ()
> +  "Display and return the number of listens in the offline queue."
> +  (interactive)
> +  (let ((count (length (emms-listenbrainz-scrobbler--read-queue))))
> +    (message "Listenbrainz: %d listens queued for submission." count)
> +    count))
> +
> +(defun emms-listenbrainz-scrobbler-clear-queue ()
> +  "Clear all listens from the offline queue.
> +This discards listens that have not yet been submitted.  This action
> +cannot be undone."
> +  (interactive)
> +  (when (yes-or-no-p "Discard all queued offline listens? ")
> +    (emms-listenbrainz-scrobbler--write-queue nil)
> +    (message "Listenbrainz: offline queue cleared.")))
>  
>  
>  ;;* Hooks
> @@ -230,11 +527,15 @@ That is, if it has been played for 240 seconds or half the length of the track."
>  
>  
>  ;;* Entry points
> +
>  (defun emms-listenbrainz-scrobbler-enable ()
> -  "Enable the scrobbler and submit played tracks."
> +  "Enable the scrobbler and submit played tracks.
> +Also starts a periodic timer to flush any queued offline listens, and
> +performs an initial flush attempt."
>    (interactive)
>    ;; We rely on emms-playing-time so check for it!
> -  (unless emms-playing-time-mode (error "Listenbrainz scrobbler: please activate emms-playing-time-mode"))
> +  (unless emms-playing-time-mode
> +    (error "Listenbrainz scrobbler: please activate emms-playing-time-mode"))
>    ;; check we have credentials
>    (if (emms-listenbrainz-scrobbler-get-token)
>        (unless emms-listenbrainz-scrobbler-running
> @@ -244,7 +545,15 @@ That is, if it has been played for 240 seconds or half the length of the track."
>  		  'emms-listenbrainz-scrobbler-stop-hook)
>  	(add-hook 'emms-player-finished-hook
>  		  'emms-listenbrainz-scrobbler-stop-hook)
> -	(setq emms-listenbrainz-scrobbler-running t))
> +	;; Start periodic queue flushing
> +	(when emms-listenbrainz-scrobbler-flush-interval
> +	  (setq emms-listenbrainz-scrobbler-queue-timer
> +		(run-at-time t
> +			     emms-listenbrainz-scrobbler-flush-interval
> +			     #'emms-listenbrainz-scrobbler-flush-queue)))
> +	(setq emms-listenbrainz-scrobbler-running t)
> +	;; Flush any listens queued from previous sessions
> +	(emms-listenbrainz-scrobbler-flush-queue))
>      (error "Listenbrainz scrobbler: no user token.  Please supply and try again")))
>  
>  (defun emms-listenbrainz-scrobbler-disable ()
> @@ -257,6 +566,10 @@ That is, if it has been played for 240 seconds or half the length of the track."
>  		 'emms-listenbrainz-scrobbler-stop-hook)
>      (remove-hook 'emms-player-finished-hook
>  		 'emms-listenbrainz-scrobbler-stop-hook)
> +    ;; Cancel periodic flush timer
> +    (when emms-listenbrainz-scrobbler-queue-timer
> +      (cancel-timer emms-listenbrainz-scrobbler-queue-timer)
> +      (setq emms-listenbrainz-scrobbler-queue-timer nil))
>      (setq emms-listenbrainz-scrobbler-running nil)))

-- 
   "Cut your own wood and it will warm you twice"