mirror of
https://github.com/idanoo/autobrr
synced 2025-07-23 08:49:13 +00:00
Feature: Support multiline irc parsing (#39)
* feat: initial multiline support * refactor: handle multiple indexers per network * wip: setup indexer * build: add docker compose for testing * chore: remove temp mock indexers * chore: update deps * refactor: update and store network handler * build: update test compose * chore: minor cleanup
This commit is contained in:
Ludvig Lundgren
GitHub
parent
506cef6f0f
commit
c4d580eb03
17 changed files with 1100 additions and 1042 deletions
359
internal/irc/announce.go
Normal file
359
internal/irc/announce.go
Normal file
|
|
@ -0,0 +1,359 @@
|
|||
package irc
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"net/url"
|
||||
"regexp"
|
||||
"strings"
|
||||
"text/template"
|
||||
|
||||
"github.com/autobrr/autobrr/internal/domain"
|
||||
"github.com/autobrr/autobrr/internal/filter"
|
||||
"github.com/autobrr/autobrr/internal/release"
|
||||
|
||||
"github.com/rs/zerolog/log"
|
||||
)
|
||||
|
||||
type Processor interface {
|
||||
AddLineToQueue(channel string, line string) error
|
||||
}
|
||||
|
||||
type announceProcessor struct {
|
||||
indexer domain.IndexerDefinition
|
||||
|
||||
filterSvc filter.Service
|
||||
releaseSvc release.Service
|
||||
|
||||
queues map[string]chan string
|
||||
}
|
||||
|
||||
func NewAnnounceProcessor(indexer domain.IndexerDefinition, filterSvc filter.Service, releaseSvc release.Service) Processor {
|
||||
ap := &announceProcessor{
|
||||
indexer: indexer,
|
||||
filterSvc: filterSvc,
|
||||
releaseSvc: releaseSvc,
|
||||
}
|
||||
|
||||
// setup queues and consumers
|
||||
ap.setupQueues()
|
||||
ap.setupQueueConsumers()
|
||||
|
||||
return ap
|
||||
}
|
||||
|
||||
func (a *announceProcessor) setupQueues() {
|
||||
queues := make(map[string]chan string)
|
||||
for _, channel := range a.indexer.IRC.Channels {
|
||||
channel = strings.ToLower(channel)
|
||||
|
||||
queues[channel] = make(chan string, 128)
|
||||
}
|
||||
|
||||
a.queues = queues
|
||||
}
|
||||
|
||||
func (a *announceProcessor) setupQueueConsumers() {
|
||||
for queueName, queue := range a.queues {
|
||||
go func(name string, q chan string) {
|
||||
a.processQueue(q)
|
||||
}(queueName, queue)
|
||||
}
|
||||
}
|
||||
|
||||
func (a *announceProcessor) processQueue(queue chan string) {
|
||||
for {
|
||||
tmpVars := map[string]string{}
|
||||
parseFailed := false
|
||||
//patternParsed := false
|
||||
|
||||
for _, pattern := range a.indexer.Parse.Lines {
|
||||
line, err := a.getNextLine(queue)
|
||||
if err != nil {
|
||||
log.Error().Stack().Err(err).Msg("could not get line from queue")
|
||||
return
|
||||
}
|
||||
log.Trace().Msgf("announce: process line: %v", line)
|
||||
|
||||
// check should ignore
|
||||
|
||||
match, err := a.parseExtract(pattern.Pattern, pattern.Vars, tmpVars, line)
|
||||
if err != nil {
|
||||
log.Debug().Msgf("error parsing extract: %v", line)
|
||||
|
||||
parseFailed = true
|
||||
break
|
||||
}
|
||||
|
||||
if !match {
|
||||
log.Debug().Msgf("line not matching expected regex pattern: %v", line)
|
||||
parseFailed = true
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
if parseFailed {
|
||||
continue
|
||||
}
|
||||
|
||||
newRelease, err := domain.NewRelease(a.indexer.Identifier, "")
|
||||
if err != nil {
|
||||
log.Error().Err(err).Msg("could not create new release")
|
||||
break
|
||||
}
|
||||
|
||||
// on lines matched
|
||||
err = a.onLinesMatched(a.indexer, tmpVars, newRelease)
|
||||
if err != nil {
|
||||
log.Debug().Msgf("error match line: %v", "")
|
||||
break
|
||||
}
|
||||
|
||||
// send to filter service to take care of the rest
|
||||
|
||||
// find and check filter
|
||||
filterOK, foundFilter, err := a.filterSvc.FindAndCheckFilters(newRelease)
|
||||
if err != nil {
|
||||
log.Error().Err(err).Msg("could not find filter")
|
||||
break
|
||||
}
|
||||
|
||||
// no foundFilter found, lets return
|
||||
if !filterOK || foundFilter == nil {
|
||||
log.Trace().Msg("no matching filter found")
|
||||
break
|
||||
|
||||
// TODO check in config for "Save all releases"
|
||||
// Save as rejected
|
||||
//newRelease.FilterStatus = domain.ReleaseStatusFilterRejected
|
||||
//err = s.releaseSvc.Store(ctx, newRelease)
|
||||
//if err != nil {
|
||||
// log.Error().Err(err).Msgf("error writing release to database: %+v", newRelease)
|
||||
// return nil
|
||||
//}
|
||||
//return nil
|
||||
}
|
||||
|
||||
// save release
|
||||
newRelease.Filter = foundFilter
|
||||
newRelease.FilterName = foundFilter.Name
|
||||
newRelease.FilterID = foundFilter.ID
|
||||
|
||||
newRelease.FilterStatus = domain.ReleaseStatusFilterApproved
|
||||
err = a.releaseSvc.Store(context.Background(), newRelease)
|
||||
if err != nil {
|
||||
log.Error().Err(err).Msgf("error writing release to database: %+v", newRelease)
|
||||
break
|
||||
}
|
||||
|
||||
log.Info().Msgf("Matched '%v' (%v) for %v", newRelease.TorrentName, newRelease.Filter.Name, newRelease.Indexer)
|
||||
|
||||
// process release
|
||||
go func() {
|
||||
err = a.releaseSvc.Process(*newRelease)
|
||||
if err != nil {
|
||||
log.Error().Err(err).Msgf("could not process release: %+v", newRelease)
|
||||
}
|
||||
}()
|
||||
}
|
||||
}
|
||||
|
||||
func (a *announceProcessor) getNextLine(queue chan string) (string, error) {
|
||||
for {
|
||||
line, ok := <-queue
|
||||
if !ok {
|
||||
return "", errors.New("could not queue line")
|
||||
}
|
||||
|
||||
return line, nil
|
||||
}
|
||||
}
|
||||
|
||||
func (a *announceProcessor) AddLineToQueue(channel string, line string) error {
|
||||
channel = strings.ToLower(channel)
|
||||
queue, ok := a.queues[channel]
|
||||
if !ok {
|
||||
return fmt.Errorf("no queue for channel (%v) found", channel)
|
||||
}
|
||||
|
||||
queue <- line
|
||||
log.Trace().Msgf("announce: queued line: %v", line)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (a *announceProcessor) parseExtract(pattern string, vars []string, tmpVars map[string]string, line string) (bool, error) {
|
||||
|
||||
rxp, err := regExMatch(pattern, line)
|
||||
if err != nil {
|
||||
log.Debug().Msgf("did not match expected line: %v", line)
|
||||
}
|
||||
|
||||
if rxp == nil {
|
||||
return false, nil
|
||||
}
|
||||
|
||||
// extract matched
|
||||
for i, v := range vars {
|
||||
value := ""
|
||||
|
||||
if rxp[i] != "" {
|
||||
value = rxp[i]
|
||||
// tmpVars[v] = rxp[i]
|
||||
}
|
||||
|
||||
tmpVars[v] = value
|
||||
}
|
||||
return true, nil
|
||||
}
|
||||
|
||||
// onLinesMatched process vars into release
|
||||
func (a *announceProcessor) onLinesMatched(def domain.IndexerDefinition, vars map[string]string, release *domain.Release) error {
|
||||
var err error
|
||||
|
||||
err = release.MapVars(vars)
|
||||
|
||||
// FIXME is this even needed anymore?
|
||||
// canonicalize name
|
||||
//canonReleaseName := cleanReleaseName(release.TorrentName)
|
||||
//log.Trace().Msgf("canonicalize release name: %v", canonReleaseName)
|
||||
|
||||
// parse fields
|
||||
err = release.Parse()
|
||||
if err != nil {
|
||||
log.Error().Err(err).Msg("announce: could not parse release")
|
||||
return err
|
||||
}
|
||||
|
||||
// generate torrent url
|
||||
torrentUrl, err := a.processTorrentUrl(def.Parse.Match.TorrentURL, vars, def.SettingsMap, def.Parse.Match.Encode)
|
||||
if err != nil {
|
||||
log.Error().Err(err).Msg("announce: could not process torrent url")
|
||||
return err
|
||||
}
|
||||
|
||||
if torrentUrl != "" {
|
||||
release.TorrentURL = torrentUrl
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (a *announceProcessor) processTorrentUrl(match string, vars map[string]string, extraVars map[string]string, encode []string) (string, error) {
|
||||
tmpVars := map[string]string{}
|
||||
|
||||
// copy vars to new tmp map
|
||||
for k, v := range vars {
|
||||
tmpVars[k] = v
|
||||
}
|
||||
|
||||
// merge extra vars with vars
|
||||
if extraVars != nil {
|
||||
for k, v := range extraVars {
|
||||
tmpVars[k] = v
|
||||
}
|
||||
}
|
||||
|
||||
// handle url encode of values
|
||||
if encode != nil {
|
||||
for _, e := range encode {
|
||||
if v, ok := tmpVars[e]; ok {
|
||||
// url encode value
|
||||
t := url.QueryEscape(v)
|
||||
tmpVars[e] = t
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// setup text template to inject variables into
|
||||
tmpl, err := template.New("torrenturl").Parse(match)
|
||||
if err != nil {
|
||||
log.Error().Err(err).Msg("could not create torrent url template")
|
||||
return "", err
|
||||
}
|
||||
|
||||
var b bytes.Buffer
|
||||
err = tmpl.Execute(&b, &tmpVars)
|
||||
if err != nil {
|
||||
log.Error().Err(err).Msg("could not write torrent url template output")
|
||||
return "", err
|
||||
}
|
||||
|
||||
return b.String(), nil
|
||||
}
|
||||
|
||||
func split(r rune) bool {
|
||||
return r == ' ' || r == '.'
|
||||
}
|
||||
|
||||
func Splitter(s string, splits string) []string {
|
||||
m := make(map[rune]int)
|
||||
for _, r := range splits {
|
||||
m[r] = 1
|
||||
}
|
||||
|
||||
splitter := func(r rune) bool {
|
||||
return m[r] == 1
|
||||
}
|
||||
|
||||
return strings.FieldsFunc(s, splitter)
|
||||
}
|
||||
|
||||
func canonicalizeString(s string) []string {
|
||||
//a := strings.FieldsFunc(s, split)
|
||||
a := Splitter(s, " .")
|
||||
|
||||
return a
|
||||
}
|
||||
|
||||
func cleanReleaseName(input string) string {
|
||||
// Make a Regex to say we only want letters and numbers
|
||||
reg, err := regexp.Compile("[^a-zA-Z0-9]+")
|
||||
if err != nil {
|
||||
//log.Fatal(err)
|
||||
}
|
||||
processedString := reg.ReplaceAllString(input, " ")
|
||||
|
||||
return processedString
|
||||
}
|
||||
|
||||
func removeElement(s []string, i int) ([]string, error) {
|
||||
// s is [1,2,3,4,5,6], i is 2
|
||||
|
||||
// perform bounds checking first to prevent a panic!
|
||||
if i >= len(s) || i < 0 {
|
||||
return nil, fmt.Errorf("Index is out of range. Index is %d with slice length %d", i, len(s))
|
||||
}
|
||||
|
||||
// This creates a new slice by creating 2 slices from the original:
|
||||
// s[:i] -> [1, 2]
|
||||
// s[i+1:] -> [4, 5, 6]
|
||||
// and joining them together using `append`
|
||||
return append(s[:i], s[i+1:]...), nil
|
||||
}
|
||||
|
||||
func regExMatch(pattern string, value string) ([]string, error) {
|
||||
|
||||
rxp, err := regexp.Compile(pattern)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
//return errors.Wrapf(err, "invalid regex: %s", value)
|
||||
}
|
||||
|
||||
matches := rxp.FindStringSubmatch(value)
|
||||
if matches == nil {
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
res := make([]string, 0)
|
||||
if matches != nil {
|
||||
res, err = removeElement(matches, 0)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
|
||||
return res, nil
|
||||
}
|
||||
|
|
@ -10,8 +10,9 @@ import (
|
|||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/autobrr/autobrr/internal/announce"
|
||||
"github.com/autobrr/autobrr/internal/domain"
|
||||
"github.com/autobrr/autobrr/internal/filter"
|
||||
"github.com/autobrr/autobrr/internal/release"
|
||||
|
||||
"github.com/rs/zerolog/log"
|
||||
"gopkg.in/irc.v3"
|
||||
|
|
@ -21,9 +22,20 @@ var (
|
|||
connectTimeout = 15 * time.Second
|
||||
)
|
||||
|
||||
type channelHealth struct {
|
||||
name string
|
||||
monitoring bool
|
||||
monitoringSince time.Time
|
||||
lastPing time.Time
|
||||
lastAnnounce time.Time
|
||||
}
|
||||
|
||||
type Handler struct {
|
||||
network *domain.IrcNetwork
|
||||
announceService announce.Service
|
||||
network *domain.IrcNetwork
|
||||
filterService filter.Service
|
||||
releaseService release.Service
|
||||
announceProcessors map[string]Processor
|
||||
definitions []domain.IndexerDefinition
|
||||
|
||||
client *irc.Client
|
||||
conn net.Conn
|
||||
|
|
@ -33,17 +45,42 @@ type Handler struct {
|
|||
|
||||
lastPing time.Time
|
||||
lastAnnounce time.Time
|
||||
|
||||
validAnnouncers map[string]struct{}
|
||||
channels map[string]channelHealth
|
||||
}
|
||||
|
||||
func NewHandler(network domain.IrcNetwork, announceService announce.Service) *Handler {
|
||||
return &Handler{
|
||||
client: nil,
|
||||
conn: nil,
|
||||
ctx: nil,
|
||||
stopped: make(chan struct{}),
|
||||
network: &network,
|
||||
announceService: announceService,
|
||||
func NewHandler(network domain.IrcNetwork, filterService filter.Service, releaseService release.Service, definitions []domain.IndexerDefinition) *Handler {
|
||||
h := &Handler{
|
||||
client: nil,
|
||||
conn: nil,
|
||||
ctx: nil,
|
||||
stopped: make(chan struct{}),
|
||||
network: &network,
|
||||
filterService: filterService,
|
||||
releaseService: releaseService,
|
||||
definitions: definitions,
|
||||
announceProcessors: map[string]Processor{},
|
||||
validAnnouncers: map[string]struct{}{},
|
||||
}
|
||||
|
||||
// Networks can be shared by multiple indexers but channels are unique
|
||||
// so let's add a new AnnounceProcessor per channel
|
||||
for _, definition := range definitions {
|
||||
// indexers can use multiple channels, but it's not common, but let's handle that anyway.
|
||||
for _, channel := range definition.IRC.Channels {
|
||||
channel = strings.ToLower(channel)
|
||||
|
||||
h.announceProcessors[channel] = NewAnnounceProcessor(definition, filterService, releaseService)
|
||||
}
|
||||
|
||||
// create map of valid announcers
|
||||
for _, announcer := range definition.IRC.Announcers {
|
||||
h.validAnnouncers[announcer] = struct{}{}
|
||||
}
|
||||
}
|
||||
|
||||
return h
|
||||
}
|
||||
|
||||
func (s *Handler) Run() error {
|
||||
|
|
@ -101,11 +138,14 @@ func (s *Handler) Run() error {
|
|||
switch m.Command {
|
||||
case "001":
|
||||
// 001 is a welcome event, so we join channels there
|
||||
err := s.onConnect(c, s.network.Channels)
|
||||
err := s.onConnect(s.network.Channels)
|
||||
if err != nil {
|
||||
log.Error().Msgf("error joining channels %v", err)
|
||||
}
|
||||
|
||||
case "372", "375", "376":
|
||||
// Handle MOTD
|
||||
|
||||
// 322 TOPIC
|
||||
// 333 UP
|
||||
// 353 @
|
||||
|
|
@ -189,6 +229,10 @@ func (s *Handler) GetNetwork() *domain.IrcNetwork {
|
|||
return s.network
|
||||
}
|
||||
|
||||
func (s *Handler) UpdateNetwork(network *domain.IrcNetwork) {
|
||||
s.network = network
|
||||
}
|
||||
|
||||
func (s *Handler) Stop() {
|
||||
s.cancel()
|
||||
|
||||
|
|
@ -210,7 +254,23 @@ func (s *Handler) isStopped() bool {
|
|||
}
|
||||
}
|
||||
|
||||
func (s *Handler) onConnect(client *irc.Client, channels []domain.IrcChannel) error {
|
||||
func (s *Handler) Restart() error {
|
||||
s.cancel()
|
||||
|
||||
if !s.isStopped() {
|
||||
close(s.stopped)
|
||||
}
|
||||
|
||||
if s.conn != nil {
|
||||
s.conn.Close()
|
||||
}
|
||||
|
||||
time.Sleep(2 * time.Second)
|
||||
|
||||
return s.Run()
|
||||
}
|
||||
|
||||
func (s *Handler) onConnect(channels []domain.IrcChannel) error {
|
||||
identified := false
|
||||
|
||||
time.Sleep(2 * time.Second)
|
||||
|
|
@ -239,7 +299,7 @@ func (s *Handler) onConnect(client *irc.Client, channels []domain.IrcChannel) er
|
|||
|
||||
if !identified {
|
||||
for _, channel := range channels {
|
||||
err := s.handleJoinChannel(channel.Name)
|
||||
err := s.HandleJoinChannel(channel.Name, channel.Password)
|
||||
if err != nil {
|
||||
log.Error().Err(err)
|
||||
return err
|
||||
|
|
@ -255,36 +315,47 @@ func (s *Handler) OnJoin(msg string) (interface{}, error) {
|
|||
}
|
||||
|
||||
func (s *Handler) onMessage(msg *irc.Message) error {
|
||||
//log.Debug().Msgf("raw msg: %v", msg)
|
||||
|
||||
// check if message is from announce bot and correct channel, if not return
|
||||
//if msg.Name != s.network. {
|
||||
//
|
||||
//}
|
||||
|
||||
// parse announce
|
||||
channel := &msg.Params[0]
|
||||
announcer := &msg.Name
|
||||
message := msg.Trailing()
|
||||
// TODO add network
|
||||
|
||||
// add correlationID and tracing
|
||||
|
||||
announceID := fmt.Sprintf("%v:%v:%v", s.network.Server, *channel, *announcer)
|
||||
announceID = strings.ToLower(announceID)
|
||||
// check if message is from announce bot, if not return
|
||||
validAnnouncer := s.isValidAnnouncer(*announcer)
|
||||
if !validAnnouncer {
|
||||
return nil
|
||||
}
|
||||
|
||||
// clean message
|
||||
cleanedMsg := cleanMessage(message)
|
||||
log.Debug().Msgf("%v: %v %v: %v", s.network.Server, *channel, *announcer, cleanedMsg)
|
||||
|
||||
s.lastAnnounce = time.Now()
|
||||
s.setLastAnnounce()
|
||||
|
||||
go func() {
|
||||
err := s.announceService.Parse(announceID, cleanedMsg)
|
||||
if err != nil {
|
||||
log.Error().Err(err).Msgf("could not parse line: %v", cleanedMsg)
|
||||
}
|
||||
}()
|
||||
if err := s.sendToAnnounceProcessor(*channel, cleanedMsg); err != nil {
|
||||
log.Error().Stack().Err(err).Msgf("could not queue line: %v", cleanedMsg)
|
||||
return err
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *Handler) sendToAnnounceProcessor(channel string, msg string) error {
|
||||
channel = strings.ToLower(channel)
|
||||
|
||||
// check if queue exists
|
||||
queue, ok := s.announceProcessors[channel]
|
||||
if !ok {
|
||||
return fmt.Errorf("queue '%v' not found", channel)
|
||||
}
|
||||
|
||||
// if it exists, add msg
|
||||
err := queue.AddLineToQueue(channel, msg)
|
||||
if err != nil {
|
||||
log.Error().Stack().Err(err).Msgf("could not queue line: %v", msg)
|
||||
return err
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
|
@ -302,13 +373,19 @@ func (s *Handler) sendPrivMessage(msg string) error {
|
|||
return nil
|
||||
}
|
||||
|
||||
func (s *Handler) handleJoinChannel(channel string) error {
|
||||
m := irc.Message{
|
||||
Command: "JOIN",
|
||||
Params: []string{channel},
|
||||
func (s *Handler) HandleJoinChannel(channel string, password string) error {
|
||||
// support channel password
|
||||
params := []string{channel}
|
||||
if password != "" {
|
||||
params = append(params, password)
|
||||
}
|
||||
|
||||
log.Debug().Msgf("%v: %v", s.network.Server, m.String())
|
||||
m := irc.Message{
|
||||
Command: "JOIN",
|
||||
Params: params,
|
||||
}
|
||||
|
||||
log.Trace().Msgf("%v: sending %v", s.network.Server, m.String())
|
||||
|
||||
time.Sleep(1 * time.Second)
|
||||
|
||||
|
|
@ -323,6 +400,27 @@ func (s *Handler) handleJoinChannel(channel string) error {
|
|||
return nil
|
||||
}
|
||||
|
||||
func (s *Handler) HandlePartChannel(channel string) error {
|
||||
m := irc.Message{
|
||||
Command: "PART",
|
||||
Params: []string{channel},
|
||||
}
|
||||
|
||||
log.Debug().Msgf("%v: %v", s.network.Server, m.String())
|
||||
|
||||
time.Sleep(1 * time.Second)
|
||||
|
||||
err := s.client.Write(m.String())
|
||||
if err != nil {
|
||||
log.Error().Err(err).Msgf("error handling part: %v", m.String())
|
||||
return err
|
||||
}
|
||||
|
||||
log.Info().Msgf("Left channel '%v' on network '%s'", channel, s.network.Server)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *Handler) handleJoined(msg *irc.Message) {
|
||||
log.Debug().Msgf("%v: JOINED: %v", s.network.Server, msg.Trailing())
|
||||
|
||||
|
|
@ -384,6 +482,40 @@ func (s *Handler) handleNickServPRIVMSG(nick, password string) error {
|
|||
return nil
|
||||
}
|
||||
|
||||
func (s *Handler) HandleNickServIdentify(nick, password string) error {
|
||||
m := irc.Message{
|
||||
Command: "PRIVMSG",
|
||||
Params: []string{"NickServ", "IDENTIFY", nick, password},
|
||||
}
|
||||
|
||||
log.Debug().Msgf("%v: NickServ: %v", s.network.Server, m.String())
|
||||
|
||||
err := s.client.Write(m.String())
|
||||
if err != nil {
|
||||
log.Error().Stack().Err(err).Msgf("error identifying with nickserv: %v", m.String())
|
||||
return err
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *Handler) HandleNickChange(nick string) error {
|
||||
m := irc.Message{
|
||||
Command: "NICK",
|
||||
Params: []string{nick},
|
||||
}
|
||||
|
||||
log.Debug().Msgf("%v: Nick change: %v", s.network.Server, m.String())
|
||||
|
||||
err := s.client.Write(m.String())
|
||||
if err != nil {
|
||||
log.Error().Stack().Err(err).Msgf("error changing nick: %v", m.String())
|
||||
return err
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *Handler) handleMode(msg *irc.Message) error {
|
||||
log.Debug().Msgf("%v: MODE: %v %v", s.network.Server, msg.User, msg.Trailing())
|
||||
|
||||
|
|
@ -395,7 +527,7 @@ func (s *Handler) handleMode(msg *irc.Message) error {
|
|||
}
|
||||
|
||||
for _, ch := range s.network.Channels {
|
||||
err := s.handleJoinChannel(ch.Name)
|
||||
err := s.HandleJoinChannel(ch.Name, ch.Password)
|
||||
if err != nil {
|
||||
log.Error().Err(err).Msgf("error joining channel: %v", ch.Name)
|
||||
continue
|
||||
|
|
@ -423,11 +555,36 @@ func (s *Handler) handlePing(msg *irc.Message) error {
|
|||
return err
|
||||
}
|
||||
|
||||
s.lastPing = time.Now()
|
||||
s.setLastPing()
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *Handler) isValidAnnouncer(nick string) bool {
|
||||
_, ok := s.validAnnouncers[nick]
|
||||
if !ok {
|
||||
return false
|
||||
}
|
||||
|
||||
return true
|
||||
}
|
||||
|
||||
func (s *Handler) setLastAnnounce() {
|
||||
s.lastAnnounce = time.Now()
|
||||
}
|
||||
|
||||
func (s *Handler) GetLastAnnounce() time.Time {
|
||||
return s.lastAnnounce
|
||||
}
|
||||
|
||||
func (s *Handler) setLastPing() {
|
||||
s.lastPing = time.Now()
|
||||
}
|
||||
|
||||
func (s *Handler) GetLastPing() time.Time {
|
||||
return s.lastPing
|
||||
}
|
||||
|
||||
// irc line can contain lots of extra stuff like color so lets clean that
|
||||
func cleanMessage(message string) string {
|
||||
var regexMessageClean = `\x0f|\x1f|\x02|\x03(?:[\d]{1,2}(?:,[\d]{1,2})?)?`
|
||||
|
|
|
|||
|
|
@ -5,8 +5,10 @@ import (
|
|||
"fmt"
|
||||
"sync"
|
||||
|
||||
"github.com/autobrr/autobrr/internal/announce"
|
||||
"github.com/autobrr/autobrr/internal/domain"
|
||||
"github.com/autobrr/autobrr/internal/filter"
|
||||
"github.com/autobrr/autobrr/internal/indexer"
|
||||
"github.com/autobrr/autobrr/internal/release"
|
||||
|
||||
"github.com/rs/zerolog/log"
|
||||
)
|
||||
|
|
@ -18,30 +20,35 @@ type Service interface {
|
|||
ListNetworks(ctx context.Context) ([]domain.IrcNetwork, error)
|
||||
GetNetworkByID(id int64) (*domain.IrcNetwork, error)
|
||||
DeleteNetwork(ctx context.Context, id int64) error
|
||||
StoreNetwork(network *domain.IrcNetwork) error
|
||||
StoreNetwork(ctx context.Context, network *domain.IrcNetwork) error
|
||||
UpdateNetwork(ctx context.Context, network *domain.IrcNetwork) error
|
||||
StoreChannel(networkID int64, channel *domain.IrcChannel) error
|
||||
}
|
||||
|
||||
type service struct {
|
||||
repo domain.IrcRepo
|
||||
announceService announce.Service
|
||||
indexerMap map[string]string
|
||||
handlers map[string]*Handler
|
||||
repo domain.IrcRepo
|
||||
filterService filter.Service
|
||||
indexerService indexer.Service
|
||||
releaseService release.Service
|
||||
indexerMap map[string]string
|
||||
handlers map[string]*Handler
|
||||
|
||||
stopWG sync.WaitGroup
|
||||
lock sync.Mutex
|
||||
}
|
||||
|
||||
func NewService(repo domain.IrcRepo, announceService announce.Service) Service {
|
||||
func NewService(repo domain.IrcRepo, filterService filter.Service, indexerSvc indexer.Service, releaseSvc release.Service) Service {
|
||||
return &service{
|
||||
repo: repo,
|
||||
announceService: announceService,
|
||||
handlers: make(map[string]*Handler),
|
||||
repo: repo,
|
||||
filterService: filterService,
|
||||
indexerService: indexerSvc,
|
||||
releaseService: releaseSvc,
|
||||
handlers: make(map[string]*Handler),
|
||||
}
|
||||
}
|
||||
|
||||
func (s *service) StartHandlers() {
|
||||
networks, err := s.repo.ListNetworks(context.Background())
|
||||
networks, err := s.repo.FindActiveNetworks(context.Background())
|
||||
if err != nil {
|
||||
log.Error().Msgf("failed to list networks: %v", err)
|
||||
}
|
||||
|
|
@ -61,9 +68,14 @@ func (s *service) StartHandlers() {
|
|||
}
|
||||
network.Channels = channels
|
||||
|
||||
handler := NewHandler(network, s.announceService)
|
||||
// find indexer definitions for network and add
|
||||
definitions := s.indexerService.GetIndexersByIRCNetwork(network.Server)
|
||||
|
||||
s.handlers[network.Name] = handler
|
||||
// init new irc handler
|
||||
handler := NewHandler(network, s.filterService, s.releaseService, definitions)
|
||||
|
||||
// TODO use network.Server? + nick? Need a way to use multiple indexers for one network if same nick
|
||||
s.handlers[network.Server] = handler
|
||||
s.lock.Unlock()
|
||||
|
||||
log.Debug().Msgf("starting network: %+v", network.Name)
|
||||
|
|
@ -91,23 +103,33 @@ func (s *service) StopHandlers() {
|
|||
|
||||
func (s *service) startNetwork(network domain.IrcNetwork) error {
|
||||
// look if we have the network in handlers already, if so start it
|
||||
if handler, found := s.handlers[network.Name]; found {
|
||||
if existingHandler, found := s.handlers[network.Server]; found {
|
||||
log.Debug().Msgf("starting network: %+v", network.Name)
|
||||
|
||||
if handler.conn != nil {
|
||||
if existingHandler.conn != nil {
|
||||
go func() {
|
||||
if err := handler.Run(); err != nil {
|
||||
log.Error().Err(err).Msgf("failed to start handler for network %q", handler.network.Name)
|
||||
if err := existingHandler.Run(); err != nil {
|
||||
log.Error().Err(err).Msgf("failed to start existingHandler for network %q", existingHandler.network.Name)
|
||||
}
|
||||
}()
|
||||
}
|
||||
} else {
|
||||
// if not found in handlers, lets add it and run it
|
||||
|
||||
handler := NewHandler(network, s.announceService)
|
||||
|
||||
s.lock.Lock()
|
||||
s.handlers[network.Name] = handler
|
||||
channels, err := s.repo.ListChannels(network.ID)
|
||||
if err != nil {
|
||||
log.Error().Err(err).Msgf("failed to list channels for network %q", network.Server)
|
||||
}
|
||||
network.Channels = channels
|
||||
|
||||
// find indexer definitions for network and add
|
||||
definitions := s.indexerService.GetIndexersByIRCNetwork(network.Server)
|
||||
|
||||
// init new irc handler
|
||||
handler := NewHandler(network, s.filterService, s.releaseService, definitions)
|
||||
|
||||
s.handlers[network.Server] = handler
|
||||
s.lock.Unlock()
|
||||
|
||||
log.Debug().Msgf("starting network: %+v", network.Name)
|
||||
|
|
@ -126,6 +148,143 @@ func (s *service) startNetwork(network domain.IrcNetwork) error {
|
|||
return nil
|
||||
}
|
||||
|
||||
func (s *service) checkIfNetworkRestartNeeded(network *domain.IrcNetwork) error {
|
||||
// look if we have the network in handlers, if so restart it
|
||||
// TODO check if we need to add indexerDefinitions etc
|
||||
if existingHandler, found := s.handlers[network.Server]; found {
|
||||
log.Debug().Msgf("decide if irc network handler needs restart or updating: %+v", network.Name)
|
||||
|
||||
// if server, tls, invite command, port : changed - restart
|
||||
// if nickserv account, nickserv password : changed - stay connected, and change those
|
||||
// if channels len : changes - join or leave
|
||||
if existingHandler.conn != nil {
|
||||
handler := existingHandler.GetNetwork()
|
||||
restartNeeded := false
|
||||
|
||||
if handler.Server != network.Server {
|
||||
restartNeeded = true
|
||||
} else if handler.Port != network.Port {
|
||||
restartNeeded = true
|
||||
} else if handler.TLS != network.TLS {
|
||||
restartNeeded = true
|
||||
} else if handler.InviteCommand != network.InviteCommand {
|
||||
restartNeeded = true
|
||||
}
|
||||
if restartNeeded {
|
||||
log.Info().Msgf("irc: restarting network: %+v", network.Name)
|
||||
|
||||
// we need to reinitialize with new network config
|
||||
existingHandler.UpdateNetwork(network)
|
||||
|
||||
go func() {
|
||||
if err := existingHandler.Restart(); err != nil {
|
||||
log.Error().Stack().Err(err).Msgf("failed to restart network %q", existingHandler.network.Name)
|
||||
}
|
||||
}()
|
||||
|
||||
// return now since the restart will read the network again OR FIXME
|
||||
return nil
|
||||
}
|
||||
|
||||
if handler.NickServ.Account != network.NickServ.Account {
|
||||
log.Debug().Msg("changing nick")
|
||||
|
||||
err := existingHandler.HandleNickChange(network.NickServ.Account)
|
||||
if err != nil {
|
||||
log.Error().Stack().Err(err).Msgf("failed to change nick %q", network.NickServ.Account)
|
||||
}
|
||||
} else if handler.NickServ.Password != network.NickServ.Password {
|
||||
log.Debug().Msg("nickserv: changing password")
|
||||
|
||||
err := existingHandler.HandleNickServIdentify(network.NickServ.Account, network.NickServ.Password)
|
||||
if err != nil {
|
||||
log.Error().Stack().Err(err).Msgf("failed to identify with nickserv %q", network.NickServ.Account)
|
||||
}
|
||||
}
|
||||
|
||||
// join or leave channels
|
||||
// loop over handler channels,
|
||||
var expectedChannels = make(map[string]struct{}, 0)
|
||||
var handlerChannels = make(map[string]struct{}, 0)
|
||||
var channelsToLeave = make([]string, 0)
|
||||
var channelsToJoin = make([]domain.IrcChannel, 0)
|
||||
|
||||
// create map of expected channels
|
||||
for _, channel := range network.Channels {
|
||||
expectedChannels[channel.Name] = struct{}{}
|
||||
}
|
||||
|
||||
// check current channels of handler against expected
|
||||
for _, handlerChan := range handler.Channels {
|
||||
handlerChannels[handlerChan.Name] = struct{}{}
|
||||
|
||||
_, ok := expectedChannels[handlerChan.Name]
|
||||
if ok {
|
||||
// if handler channel matches network channel next
|
||||
continue
|
||||
}
|
||||
|
||||
// if not expected, leave
|
||||
channelsToLeave = append(channelsToLeave, handlerChan.Name)
|
||||
}
|
||||
|
||||
for _, channel := range network.Channels {
|
||||
_, ok := handlerChannels[channel.Name]
|
||||
if ok {
|
||||
continue
|
||||
}
|
||||
|
||||
// if expected channel not in handler channels, add to join
|
||||
// use channel struct for extra info
|
||||
channelsToJoin = append(channelsToJoin, channel)
|
||||
}
|
||||
|
||||
// leave channels
|
||||
for _, leaveChannel := range channelsToLeave {
|
||||
err := existingHandler.HandlePartChannel(leaveChannel)
|
||||
if err != nil {
|
||||
log.Error().Stack().Err(err).Msgf("failed to leave channel: %q", leaveChannel)
|
||||
}
|
||||
}
|
||||
|
||||
// join channels
|
||||
for _, joinChannel := range channelsToJoin {
|
||||
// TODO handle invite commands before?
|
||||
err := existingHandler.HandleJoinChannel(joinChannel.Name, joinChannel.Password)
|
||||
if err != nil {
|
||||
log.Error().Stack().Err(err).Msgf("failed to join channel: %q", joinChannel.Name)
|
||||
}
|
||||
}
|
||||
}
|
||||
} else {
|
||||
err := s.startNetwork(*network)
|
||||
if err != nil {
|
||||
log.Error().Stack().Err(err).Msgf("failed to start network: %q", network.Name)
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *service) restartNetwork(network domain.IrcNetwork) error {
|
||||
// look if we have the network in handlers, if so restart it
|
||||
if existingHandler, found := s.handlers[network.Server]; found {
|
||||
log.Info().Msgf("restarting network: %+v", network.Name)
|
||||
|
||||
if existingHandler.conn != nil {
|
||||
go func() {
|
||||
if err := existingHandler.Restart(); err != nil {
|
||||
log.Error().Err(err).Msgf("failed to restart network %q", existingHandler.network.Name)
|
||||
}
|
||||
}()
|
||||
}
|
||||
}
|
||||
|
||||
// TODO handle full restart
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *service) StopNetwork(name string) error {
|
||||
if handler, found := s.handlers[name]; found {
|
||||
handler.Stop()
|
||||
|
|
@ -135,6 +294,27 @@ func (s *service) StopNetwork(name string) error {
|
|||
return nil
|
||||
}
|
||||
|
||||
func (s *service) StopAndRemoveNetwork(name string) error {
|
||||
if handler, found := s.handlers[name]; found {
|
||||
handler.Stop()
|
||||
|
||||
// remove from handlers
|
||||
delete(s.handlers, name)
|
||||
log.Debug().Msgf("stopped network: %+v", name)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *service) StopNetworkIfRunning(name string) error {
|
||||
if handler, found := s.handlers[name]; found {
|
||||
handler.Stop()
|
||||
log.Debug().Msgf("stopped network: %+v", name)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *service) GetNetworkByID(id int64) (*domain.IrcNetwork, error) {
|
||||
network, err := s.repo.GetNetworkByID(id)
|
||||
if err != nil {
|
||||
|
|
@ -184,7 +364,8 @@ func (s *service) DeleteNetwork(ctx context.Context, id int64) error {
|
|||
log.Debug().Msgf("delete network: %v", id)
|
||||
|
||||
// Remove network and handler
|
||||
if err = s.StopNetwork(network.Name); err != nil {
|
||||
//if err = s.StopNetwork(network.Server); err != nil {
|
||||
if err = s.StopAndRemoveNetwork(network.Server); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
|
|
@ -195,55 +376,91 @@ func (s *service) DeleteNetwork(ctx context.Context, id int64) error {
|
|||
return nil
|
||||
}
|
||||
|
||||
func (s *service) StoreNetwork(network *domain.IrcNetwork) error {
|
||||
if err := s.repo.StoreNetwork(network); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
log.Debug().Msgf("store network: %+v", network)
|
||||
func (s *service) UpdateNetwork(ctx context.Context, network *domain.IrcNetwork) error {
|
||||
|
||||
if network.Channels != nil {
|
||||
for _, channel := range network.Channels {
|
||||
if err := s.repo.StoreChannel(network.ID, &channel); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := s.repo.StoreNetworkChannels(ctx, network.ID, network.Channels); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
if err := s.repo.UpdateNetwork(ctx, network); err != nil {
|
||||
return err
|
||||
}
|
||||
log.Debug().Msgf("irc.service: update network: %+v", network)
|
||||
|
||||
// stop or start network
|
||||
// TODO get current state to see if enabled or not?
|
||||
if network.Enabled {
|
||||
err := s.startNetwork(*network)
|
||||
// if server, tls, invite command, port : changed - restart
|
||||
// if nickserv account, nickserv password : changed - stay connected, and change those
|
||||
// if channels len : changes - join or leave
|
||||
err := s.checkIfNetworkRestartNeeded(network)
|
||||
if err != nil {
|
||||
log.Error().Err(err).Msgf("could not start network: %+v", network.Name)
|
||||
return fmt.Errorf("could not start network: %v", network.Name)
|
||||
log.Error().Stack().Err(err).Msgf("could not restart network: %+v", network.Name)
|
||||
return fmt.Errorf("could not restart network: %v", network.Name)
|
||||
}
|
||||
|
||||
} else {
|
||||
err := s.StopNetwork(network.Name)
|
||||
// TODO take into account multiple channels per network
|
||||
//err := s.StopNetwork(network.Server)
|
||||
err := s.StopAndRemoveNetwork(network.Server)
|
||||
if err != nil {
|
||||
log.Error().Err(err).Msgf("could not stop network: %+v", network.Name)
|
||||
log.Error().Stack().Err(err).Msgf("could not stop network: %+v", network.Name)
|
||||
return fmt.Errorf("could not stop network: %v", network.Name)
|
||||
}
|
||||
}
|
||||
|
||||
// stop or start network
|
||||
//if !network.Enabled {
|
||||
// log.Debug().Msgf("stopping network: %+v", network.Name)
|
||||
//
|
||||
// err := s.StopNetwork(network.Name)
|
||||
// if err != nil {
|
||||
// log.Error().Err(err).Msgf("could not stop network: %+v", network.Name)
|
||||
// return fmt.Errorf("could not stop network: %v", network.Name)
|
||||
// }
|
||||
//} else {
|
||||
// log.Debug().Msgf("starting network: %+v", network.Name)
|
||||
//
|
||||
// err := s.startNetwork(*network)
|
||||
// if err != nil {
|
||||
// log.Error().Err(err).Msgf("could not start network: %+v", network.Name)
|
||||
// return fmt.Errorf("could not start network: %v", network.Name)
|
||||
// }
|
||||
//}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *service) StoreNetwork(ctx context.Context, network *domain.IrcNetwork) error {
|
||||
existingNetwork, err := s.repo.CheckExistingNetwork(ctx, network)
|
||||
if err != nil {
|
||||
log.Error().Err(err).Msg("could not check for existing network")
|
||||
return err
|
||||
}
|
||||
|
||||
if existingNetwork == nil {
|
||||
if err := s.repo.StoreNetwork(network); err != nil {
|
||||
return err
|
||||
}
|
||||
log.Debug().Msgf("store network: %+v", network)
|
||||
|
||||
if network.Channels != nil {
|
||||
for _, channel := range network.Channels {
|
||||
if err := s.repo.StoreChannel(network.ID, &channel); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
if network.Channels != nil {
|
||||
for _, channel := range network.Channels {
|
||||
// TODO store or add. Make sure it doesn't delete before
|
||||
if err := s.repo.StoreChannel(existingNetwork.ID, &channel); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
// append channels to existing network
|
||||
network.Channels = append(network.Channels, existingNetwork.Channels...)
|
||||
}
|
||||
|
||||
if existingNetwork.Enabled {
|
||||
// if server, tls, invite command, port : changed - restart
|
||||
// if nickserv account, nickserv password : changed - stay connected, and change those
|
||||
// if channels len : changes - join or leave
|
||||
|
||||
err := s.checkIfNetworkRestartNeeded(network)
|
||||
if err != nil {
|
||||
log.Error().Err(err).Msgf("could not restart network: %+v", network.Name)
|
||||
return fmt.Errorf("could not restart network: %v", network.Name)
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue