feat(irc): manually re-process announces (#1419)

* fix(releases): add manual processing

* feat(irc): add re-process button to channel msg

* feat(irc): add missing client method

* feat(web): change reprocess icon placement

---------

Co-authored-by: martylukyy <35452459+martylukyy@users.noreply.github.com>

cmd/autobrr/main.go

@@ -8,7 +8,6 @@ import (
 	"os/signal"
 	"syscall"
 	_ "time/tzdata"
-	_ "go.uber.org/automaxprocs"
 
 	"github.com/autobrr/autobrr/internal/action"
 	"github.com/autobrr/autobrr/internal/api"
@@ -31,8 +30,11 @@ import (
 	"github.com/autobrr/autobrr/internal/user"
 
 	"github.com/asaskevich/EventBus"
+	"github.com/dcarbone/zadapters/zstdlog"
 	"github.com/r3labs/sse/v2"
+	"github.com/rs/zerolog"
 	"github.com/spf13/pflag"
+	"go.uber.org/automaxprocs/maxprocs"
 )
 
 var (
@@ -52,6 +54,13 @@ func main() {
 	// init new logger
 	log := logger.New(cfg.Config)
 
+	// Set GOMAXPROCS to match the Linux container CPU quota (if any)
+	undo, err := maxprocs.Set(maxprocs.Logger(zstdlog.NewStdLoggerWithLevel(log.With().Logger(), zerolog.InfoLevel).Printf))
+	defer undo()
+	if err != nil {
+		log.Fatal().Err(err).Msg("failed to set GOMAXPROCS")
+	}
+
 	// init dynamic config
 	cfg.DynamicReload(log)
 
@@ -106,7 +115,7 @@ func main() {
 		actionService         = action.NewService(log, actionRepo, downloadClientService, bus)
 		indexerService        = indexer.NewService(log, cfg.Config, indexerRepo, indexerAPIService, schedulingService)
 		filterService         = filter.NewService(log, filterRepo, actionRepo, releaseRepo, indexerAPIService, indexerService)
-		releaseService        = release.NewService(log, releaseRepo, actionService, filterService)
+		releaseService        = release.NewService(log, releaseRepo, actionService, filterService, indexerService)
 		ircService            = irc.NewService(log, serverEvents, ircRepo, releaseService, indexerService, notificationService)
 		feedService           = feed.NewService(log, feedRepo, feedCacheRepo, releaseService, schedulingService)
 	)

internal/domain/irc.go

@@ -94,6 +94,14 @@ type ChannelHealth struct {
 	LastAnnounce    time.Time `json:"last_announce"`
 }
 
+type IRCManualProcessRequest struct {
+	NetworkId int64  `json:"-"`
+	Server    string `json:"server"`
+	Channel   string `json:"channel"`
+	Nick      string `json:"nick"`
+	Message   string `json:"msg"`
+}
+
 type SendIrcCmdRequest struct {
 	NetworkId int64  `json:"network_id"`
 	Server    string `json:"server"`

internal/domain/release.go

@@ -270,6 +270,12 @@ type ReleaseActionRetryReq struct {
 	ActionId       int
 }
 
+type ReleaseProcessReq struct {
+	IndexerIdentifier     string   `json:"indexer_identifier"`
+	IndexerImplementation string   `json:"indexer_implementation"`
+	AnnounceLines         []string `json:"announce_lines"`
+}
+
 type GetReleaseRequest struct {
 	Id int
 }

internal/http/irc.go

@@ -6,8 +6,10 @@ package http
 import (
 	"context"
 	"encoding/json"
+	"fmt"
 	"net/http"
 	"strconv"
+	"strings"
 
 	"github.com/autobrr/autobrr/internal/domain"
 
@@ -25,6 +27,7 @@ type ircService interface {
 	StoreChannel(ctx context.Context, networkID int64, channel *domain.IrcChannel) error
 	RestartNetwork(ctx context.Context, id int64) error
 	SendCmd(ctx context.Context, req *domain.SendIrcCmdRequest) error
+	ManualProcessAnnounce(ctx context.Context, req *domain.IRCManualProcessRequest) error
 }
 
 type ircHandler struct {
@@ -54,6 +57,8 @@ func (h ircHandler) Routes(r chi.Router) {
 		r.Post("/cmd", h.sendCmd)
 		r.Post("/channel", h.storeChannel)
 		r.Get("/restart", h.restartNetwork)
+
+		r.Post("/channel/{channel}/announce/process", h.announceProcess)
 	})
 
 	r.HandleFunc("/events", func(w http.ResponseWriter, r *http.Request) {
@@ -185,6 +190,58 @@ func (h ircHandler) sendCmd(w http.ResponseWriter, r *http.Request) {
 	h.encoder.NoContent(w)
 }
 
+// announceProcess manually trigger announce process
+func (h ircHandler) announceProcess(w http.ResponseWriter, r *http.Request) {
+	var (
+		ctx  = r.Context()
+		data domain.IRCManualProcessRequest
+	)
+
+	paramNetworkID := chi.URLParam(r, "networkID")
+	if paramNetworkID == "" {
+		h.encoder.StatusResponse(w, http.StatusBadRequest, map[string]interface{}{
+			"code":    "BAD_REQUEST_PARAMS",
+			"message": "parameter networkID missing",
+		})
+		return
+	}
+
+	networkID, err := strconv.Atoi(paramNetworkID)
+	if err != nil {
+		h.encoder.Error(w, err)
+		return
+	}
+
+	paramChannel := chi.URLParam(r, "channel")
+	if paramChannel == "" {
+		h.encoder.StatusResponse(w, http.StatusBadRequest, map[string]interface{}{
+			"code":    "BAD_REQUEST_PARAMS",
+			"message": "parameter channel missing",
+		})
+		return
+	}
+
+	if err := json.NewDecoder(r.Body).Decode(&data); err != nil {
+		h.encoder.Error(w, err)
+		return
+	}
+
+	data.NetworkId = int64(networkID)
+	data.Channel = paramChannel
+
+	// we cant pass # as an url parameter so the frontend has to strip it
+	if !strings.HasPrefix("#", data.Channel) {
+		data.Channel = fmt.Sprintf("#%s", data.Channel)
+	}
+
+	if err := h.service.ManualProcessAnnounce(ctx, &data); err != nil {
+		h.encoder.Error(w, err)
+		return
+	}
+
+	h.encoder.NoContent(w)
+}
+
 func (h ircHandler) storeChannel(w http.ResponseWriter, r *http.Request) {
 	var (
 		ctx       = r.Context()

internal/http/release.go

@@ -5,6 +5,7 @@ package http
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
 	"net/http"
 	"net/url"
@@ -22,6 +23,7 @@ type releaseService interface {
 	Stats(ctx context.Context) (*domain.ReleaseStats, error)
 	Delete(ctx context.Context, req *domain.DeleteReleaseRequest) error
 	Retry(ctx context.Context, req *domain.ReleaseActionRetryReq) error
+	ProcessManual(ctx context.Context, req *domain.ReleaseProcessReq) error
 }
 
 type releaseHandler struct {
@@ -43,6 +45,8 @@ func (h releaseHandler) Routes(r chi.Router) {
 	r.Get("/indexers", h.getIndexerOptions)
 	r.Delete("/", h.deleteReleases)
 
+	r.Post("/process", h.retryAction)
+
 	r.Route("/{releaseId}", func(r chi.Router) {
 		r.Post("/actions/{actionStatusId}/retry", h.retryAction)
 	})
@@ -215,6 +219,38 @@ func (h releaseHandler) deleteReleases(w http.ResponseWriter, r *http.Request) {
 	h.encoder.NoContent(w)
 }
 
+func (h releaseHandler) process(w http.ResponseWriter, r *http.Request) {
+	var req *domain.ReleaseProcessReq
+
+	err := json.NewDecoder(r.Body).Decode(&req)
+	if err != nil {
+		h.encoder.Error(w, err)
+		return
+	}
+
+	if req.IndexerIdentifier == "" {
+		h.encoder.StatusResponse(w, http.StatusBadRequest, map[string]interface{}{
+			"code":    "VALIDATION_ERROR",
+			"message": "field indexer_identifier empty",
+		})
+	}
+
+	if len(req.AnnounceLines) == 0 {
+		h.encoder.StatusResponse(w, http.StatusBadRequest, map[string]interface{}{
+			"code":    "VALIDATION_ERROR",
+			"message": "field announce_lines empty",
+		})
+	}
+
+	err = h.service.ProcessManual(r.Context(), req)
+	if err != nil {
+		h.encoder.Error(w, err)
+		return
+	}
+
+	h.encoder.NoContent(w)
+}
+
 func (h releaseHandler) retryAction(w http.ResponseWriter, r *http.Request) {
 	var (
 		req *domain.ReleaseActionRetryReq
@@ -223,7 +259,10 @@ func (h releaseHandler) retryAction(w http.ResponseWriter, r *http.Request) {
 
 	releaseIdParam := chi.URLParam(r, "releaseId")
 	if releaseIdParam == "" {
-		h.encoder.StatusError(w, http.StatusBadRequest, err)
+		h.encoder.StatusResponse(w, http.StatusBadRequest, map[string]interface{}{
+			"code":    "BAD_REQUEST_PARAMS",
+			"message": "parameter releaseId missing",
+		})
 		return
 	}
 
@@ -235,7 +274,10 @@ func (h releaseHandler) retryAction(w http.ResponseWriter, r *http.Request) {
 
 	actionStatusIdParam := chi.URLParam(r, "actionStatusId")
 	if actionStatusIdParam == "" {
-		h.encoder.StatusError(w, http.StatusBadRequest, err)
+		h.encoder.StatusResponse(w, http.StatusBadRequest, map[string]interface{}{
+			"code":    "BAD_REQUEST_PARAMS",
+			"message": "parameter actionStatusId missing",
+		})
 		return
 	}
 

internal/http/server.go

@@ -94,7 +94,7 @@ func (s Server) tryToServe(addr, protocol string) error {
 		return err
 	}
 
-	s.log.Info().Msgf("Starting server %s. Listening on %s", protocol, listener.Addr().String())
+	s.log.Info().Msgf("Starting API %s server. Listening on %s", protocol, listener.Addr().String())
 
 	server := http.Server{
 		Handler:           s.Handler(),

internal/indexer/service.go

@@ -36,6 +36,7 @@ type Service interface {
 	LoadIndexerDefinitions() error
 	GetIndexersByIRCNetwork(server string) []*domain.IndexerDefinition
 	GetTorznabIndexers() []domain.IndexerDefinition
+	GetMappedDefinitionByName(name string) (*domain.IndexerDefinition, error)
 	Start() error
 	TestApi(ctx context.Context, req domain.IndexerTestApiRequest) error
 	ToggleEnabled(ctx context.Context, indexerID int, enabled bool) error
@@ -667,6 +668,15 @@ func (s *service) getDefinitionByName(name string) *domain.IndexerDefinition {
 	return nil
 }
 
+func (s *service) GetMappedDefinitionByName(name string) (*domain.IndexerDefinition, error) {
+	v, ok := s.mappedDefinitions[name]
+	if !ok {
+		return nil, errors.New("unknown indexer identifier: %s", name)
+	}
+
+	return v, nil
+}
+
 func (s *service) getMappedDefinitionByName(name string) *domain.IndexerDefinition {
 	if v, ok := s.mappedDefinitions[name]; ok {
 		return v

internal/irc/handler.go

@@ -719,6 +719,10 @@ func (h *Handler) onMessage(msg ircmsg.Message) {
 	return
 }
 
+func (h *Handler) SendToAnnounceProcessor(channel string, msg string) error {
+	return h.sendToAnnounceProcessor(channel, msg)
+}
+
 // send the msg to announce processor
 func (h *Handler) sendToAnnounceProcessor(channel string, msg string) error {
 	channel = strings.ToLower(channel)

internal/irc/service.go

@@ -36,6 +36,7 @@ type Service interface {
 	UpdateNetwork(ctx context.Context, network *domain.IrcNetwork) error
 	StoreChannel(ctx context.Context, networkID int64, channel *domain.IrcChannel) error
 	SendCmd(ctx context.Context, req *domain.SendIrcCmdRequest) error
+	ManualProcessAnnounce(ctx context.Context, req *domain.IRCManualProcessRequest) error
 }
 
 type service struct {
@@ -408,6 +409,26 @@ func (s *service) GetNetworkByID(ctx context.Context, id int64) (*domain.IrcNetw
 	return network, nil
 }
 
+func (s *service) ManualProcessAnnounce(ctx context.Context, req *domain.IRCManualProcessRequest) error {
+	network, err := s.repo.GetNetworkByID(ctx, req.NetworkId)
+	if err != nil {
+		s.log.Error().Err(err).Msgf("failed to get network: %d", req.NetworkId)
+		return err
+	}
+
+	handler, ok := s.handlers[network.ID]
+	if !ok {
+		return errors.New("could not find irc handler with id: %d", network.ID)
+	}
+
+	err = handler.sendToAnnounceProcessor(req.Channel, req.Message)
+	if err != nil {
+		return errors.Wrap(err, "could not send manual announce to processor")
+	}
+
+	return nil
+}
+
 func (s *service) ListNetworks(ctx context.Context) ([]domain.IrcNetwork, error) {
 	networks, err := s.repo.ListNetworks(ctx)
 	if err != nil {

internal/logger/logger.go

@@ -29,6 +29,8 @@ type Logger interface {
 	With() zerolog.Context
 	RegisterSSEWriter(sse *sse.Server)
 	SetLogLevel(level string)
+	Printf(format string, v ...interface{})
+	Print(v ...interface{})
 }
 
 // DefaultLogger default logging controller
@@ -144,6 +146,18 @@ func (l *DefaultLogger) Trace() *zerolog.Event {
 	return l.log.Trace().Timestamp()
 }
 
+// Print sends a log event using debug level and no extra field.
+// Arguments are handled in the manner of fmt.Print.
+func (l *DefaultLogger) Print(v ...interface{}) {
+	l.log.Print(v...)
+}
+
+// Printf sends a log event using debug level and no extra field.
+// Arguments are handled in the manner of fmt.Printf.
+func (l *DefaultLogger) Printf(format string, v ...interface{}) {
+	l.log.Printf(format, v...)
+}
+
 // With log with context
 func (l *DefaultLogger) With() zerolog.Context {
 	return l.log.With().Timestamp()

internal/release/service.go

@@ -11,7 +11,9 @@ import (
 	"github.com/autobrr/autobrr/internal/action"
 	"github.com/autobrr/autobrr/internal/domain"
 	"github.com/autobrr/autobrr/internal/filter"
+	"github.com/autobrr/autobrr/internal/indexer"
 	"github.com/autobrr/autobrr/internal/logger"
+	"github.com/autobrr/autobrr/pkg/errors"
 
 	"github.com/rs/zerolog"
 )
@@ -28,6 +30,7 @@ type Service interface {
 	Delete(ctx context.Context, req *domain.DeleteReleaseRequest) error
 	Process(release *domain.Release)
 	ProcessMultiple(releases []*domain.Release)
+	ProcessManual(ctx context.Context, req *domain.ReleaseProcessReq) error
 	Retry(ctx context.Context, req *domain.ReleaseActionRetryReq) error
 }
 
@@ -40,16 +43,18 @@ type service struct {
 	log  zerolog.Logger
 	repo domain.ReleaseRepo
 
-	actionSvc action.Service
+	actionSvc  action.Service
-	filterSvc filter.Service
+	filterSvc  filter.Service
+	indexerSvc indexer.Service
 }
 
-func NewService(log logger.Logger, repo domain.ReleaseRepo, actionSvc action.Service, filterSvc filter.Service) Service {
+func NewService(log logger.Logger, repo domain.ReleaseRepo, actionSvc action.Service, filterSvc filter.Service, indexerSvc indexer.Service) Service {
 	return &service{
-		log:       log.With().Str("module", "release").Logger(),
+		log:        log.With().Str("module", "release").Logger(),
-		repo:      repo,
+		repo:       repo,
-		actionSvc: actionSvc,
+		actionSvc:  actionSvc,
-		filterSvc: filterSvc,
+		filterSvc:  filterSvc,
+		indexerSvc: indexerSvc,
 	}
 }
 
@@ -89,6 +94,58 @@ func (s *service) Delete(ctx context.Context, req *domain.DeleteReleaseRequest)
 	return s.repo.Delete(ctx, req)
 }
 
+func (s *service) ProcessManual(ctx context.Context, req *domain.ReleaseProcessReq) error {
+	// get indexer definition with data
+	def, err := s.indexerSvc.GetMappedDefinitionByName(req.IndexerIdentifier)
+	if err != nil {
+		return err
+	}
+
+	rls := domain.NewRelease(def.Identifier)
+
+	switch req.IndexerImplementation {
+	case string(domain.IndexerImplementationIRC):
+
+		// from announce/announce.go
+		tmpVars := map[string]string{}
+		parseFailed := false
+
+		for idx, parseLine := range def.IRC.Parse.Lines {
+			match, err := indexer.ParseLine(&s.log, parseLine.Pattern, parseLine.Vars, tmpVars, req.AnnounceLines[idx], parseLine.Ignore)
+			if err != nil {
+				parseFailed = true
+				break
+			}
+
+			if !match {
+				parseFailed = true
+				break
+			}
+		}
+
+		if parseFailed {
+			return errors.New("parse failed")
+		}
+
+		rls.Protocol = domain.ReleaseProtocol(def.Protocol)
+
+		// on lines matched
+		err = def.IRC.Parse.Parse(def, tmpVars, rls)
+		if err != nil {
+			return err
+		}
+
+	default:
+		return errors.New("implementation %q is not supported", req.IndexerImplementation)
+
+	}
+
+	// process
+	go s.Process(rls)
+
+	return nil
+}
+
 func (s *service) Process(release *domain.Release) {
 	if release == nil {
 		return

web/src/api/APIClient.ts

@@ -278,6 +278,9 @@ export const APIClient = {
     sendCmd: (cmd: SendIrcCmdRequest) => appClient.Post(`api/irc/network/${cmd.network_id}/cmd`, {
       body: cmd
     }),
+    reprocessAnnounce: (networkId: number, channel: string, msg: string) => appClient.Post(`api/irc/network/${networkId}/channel/${channel}/announce/process`, {
+      body: { msg: msg }
+    }),
     events: (network: string) => new EventSource(
       `${sseBaseUrl()}api/irc/events?stream=${encodeRFC3986URIComponent(network)}`,
       { withCredentials: true }

web/src/screens/Logs.tsx

@@ -139,7 +139,7 @@ export const Logs = () => {
               >
                 <span
                   title={entry.time}
-                  className="font-mono text-gray-500 dark:text-gray-600 mr-2 h-full"
+                  className="font-mono text-gray-500 dark:text-gray-600 h-full"
                 >
                   {format(new Date(entry.time), "HH:mm:ss")}
                 </span>
@@ -150,10 +150,10 @@ export const Logs = () => {
                       "font-mono font-semibold h-full"
                     )}
                   >
-                    {entry.level}
+                    {` ${entry.level} `}
                   </span>
                 ) : null}
-                <span className="ml-2 text-black dark:text-gray-300">
+                <span className="text-black dark:text-gray-300">
                   {entry.message}
                 </span>
               </div>

web/src/screens/settings/Irc.tsx

@@ -5,7 +5,7 @@
 
 import { Fragment, MouseEvent, useEffect, useMemo, useRef, useState } from "react";
 import { useMutation, useQueryClient, useSuspenseQuery } from "@tanstack/react-query";
-import { LockClosedIcon, LockOpenIcon, PlusIcon } from "@heroicons/react/24/solid";
+import { ArrowPathIcon, LockClosedIcon, LockOpenIcon, PlusIcon } from "@heroicons/react/24/solid";
 import { Menu, Transition } from "@headlessui/react";
 import { toast } from "react-hot-toast";
 import {
@@ -31,6 +31,7 @@ import { SettingsContext } from "@utils/Context";
 import { Checkbox } from "@components/Checkbox";
 
 import { Section } from "./_components";
+import { RingResizeSpinner } from "@components/Icons.tsx";
 
 interface SortConfig {
   key: keyof ListItemProps["network"] | "enabled";
@@ -575,6 +576,49 @@ const ListItemDropdown = ({
   );
 };
 
+interface ReprocessAnnounceProps {
+  networkId: number;
+  channel: string;
+  msg: string;
+}
+
+const ReprocessAnnounceButton = ({ networkId, channel, msg }: ReprocessAnnounceProps) => {
+  const mutation = useMutation({
+    mutationFn: (req: IrcProcessManualRequest) => APIClient.irc.reprocessAnnounce(req.network_id, req.channel, req.msg),
+    onSuccess: () => {
+      toast.custom((t) => (
+        <Toast type="success" body={`Announce sent to re-process!`} t={t} />
+      ));
+    }
+  });
+
+  const reprocessAnnounce = () => {
+    const req: IrcProcessManualRequest = {
+      network_id: networkId,
+      msg: msg,
+      channel: channel,
+    }
+
+    if (channel.startsWith("#")) {
+      req.channel = channel.replace("#", "")
+    }
+
+    mutation.mutate(req);
+  };
+
+  return (
+    <div className="block">
+    <button className="flex items-center justify-center size-5 mr-1 p-1 rounded transition border-gray-500 bg-gray-250 hover:bg-gray-300 dark:bg-gray-700 dark:hover:bg-gray-600" onClick={reprocessAnnounce} title="Re-process announce">
+      {mutation.isPending
+        ? <RingResizeSpinner className="text-blue-500 iconHeight" aria-hidden="true" />
+        : <ArrowPathIcon />
+      }
+    </button>
+    </div>
+  );
+
+}
+
 type IrcEvent = {
   channel: string;
   nick: string;
@@ -684,10 +728,16 @@ export const Events = ({ network, channel }: EventsProps) => {
             key={idx}
             className={classNames(
               settings.indentLogLines ? "grid justify-start grid-flow-col" : "",
-              settings.hideWrappedText ? "truncate hover:text-ellipsis hover:whitespace-normal" : ""
+              settings.hideWrappedText ? "truncate hover:text-ellipsis hover:whitespace-normal" : "",
+              "flex items-center hover:bg-gray-200 hover:dark:bg-gray-800"
             )}
           >
-            <span className="font-mono text-gray-500 dark:text-gray-500 mr-1"><span className="dark:text-gray-600"><span className="dark:text-gray-700">[{simplifyDate(entry.time)}]</span> {entry.nick}:</span> {entry.msg}</span>
+            <ReprocessAnnounceButton networkId={network.id} channel={channel} msg={entry.msg} />
+            <div className="flex-1">
+              <span className="font-mono text-gray-500 dark:text-gray-500 mr-1">
+                <span className="dark:text-gray-600"><span className="dark:text-gray-700">[{simplifyDate(entry.time)}]</span> {entry.nick}:</span> {entry.msg}
+              </span>
+            </div>
           </div>
         ))}
       </div>

web/src/types/Irc.d.ts

@@ -89,3 +89,10 @@ interface SendIrcCmdRequest {
   nick: string;
   msg: string;
 }
+
+interface IrcProcessManualRequest {
+  network_id: number;
+  channel: string;
+  nick?: string;
+  msg: string;
+}