Reviewed-on: #185 Co-authored-by: Tyr Mactire <tyr@pettingzoo.co> Co-committed-by: Tyr Mactire <tyr@pettingzoo.co>pull/186/head
parent
68627d7848
commit
cdaec9b3e0
Signed by:
PettingZoo Gitea
PettingZoo Gitea
GPG Key ID:
39788A4390A1372F
@ -1,8 +1,6 @@
|
||||
package scheduler
|
||||
|
||||
import (
|
||||
"git.ptzo.gdn/feditools/relay/internal/log"
|
||||
)
|
||||
import "git.ptzo.gdn/feditools/relay/internal/log"
|
||||
|
||||
type empty struct{}
|
||||
|
||||
@ -0,0 +1,210 @@
|
||||
package scheduler
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"git.ptzo.gdn/feditools/relay/cmd/relay/action"
|
||||
"git.ptzo.gdn/feditools/relay/internal/clock"
|
||||
"git.ptzo.gdn/feditools/relay/internal/config"
|
||||
"git.ptzo.gdn/feditools/relay/internal/db/bun"
|
||||
"git.ptzo.gdn/feditools/relay/internal/db/cachemem"
|
||||
"git.ptzo.gdn/feditools/relay/internal/http"
|
||||
"git.ptzo.gdn/feditools/relay/internal/kv/redis"
|
||||
"git.ptzo.gdn/feditools/relay/internal/logic/logic1"
|
||||
"git.ptzo.gdn/feditools/relay/internal/metrics"
|
||||
"git.ptzo.gdn/feditools/relay/internal/runner/asynq"
|
||||
"git.ptzo.gdn/feditools/relay/internal/token"
|
||||
"github.com/spf13/viper"
|
||||
"github.com/uptrace/uptrace-go/uptrace"
|
||||
"os"
|
||||
"os/signal"
|
||||
"syscall"
|
||||
)
|
||||
|
||||
// Start runs a worker migrations.
|
||||
var Start action.Action = func(ctx context.Context) error {
|
||||
l := logger.WithField("func", "Start")
|
||||
l.Info("starting")
|
||||
|
||||
ctx, cancel := context.WithCancel(ctx)
|
||||
|
||||
// Configure OpenTelemetry with sensible defaults.
|
||||
uptrace.ConfigureOpentelemetry(
|
||||
// copy your project DSN here or use UPTRACE_DSN env var
|
||||
//uptrace.WithDSN("https://<token>@uptrace.dev/<project_id>"),
|
||||
|
||||
uptrace.WithServiceName(viper.GetString(config.Keys.ApplicationName)),
|
||||
uptrace.WithServiceVersion(viper.GetString(config.Keys.SoftwareVersion)),
|
||||
)
|
||||
// Send buffered spans and free resources.
|
||||
defer func() {
|
||||
l.Info("closing uptrace")
|
||||
err := uptrace.Shutdown(context.Background())
|
||||
if err != nil {
|
||||
l.Errorf("closing uptrace: %s", err.Error())
|
||||
}
|
||||
}()
|
||||
|
||||
// create metrics server
|
||||
metricsServer := metrics.New(viper.GetString(config.Keys.MetricsHTTPBind))
|
||||
|
||||
// create clock module
|
||||
l.Debug("creating clock")
|
||||
clockMod := clock.NewClock()
|
||||
|
||||
// create database client
|
||||
l.Debug("creating database client")
|
||||
dbClient, err := bun.New(ctx)
|
||||
if err != nil {
|
||||
l.Errorf("db: %s", err.Error())
|
||||
cancel()
|
||||
|
||||
return err
|
||||
}
|
||||
dbCacheClient, err := cachemem.New(ctx, dbClient)
|
||||
if err != nil {
|
||||
l.Errorf("db: %s", err.Error())
|
||||
cancel()
|
||||
|
||||
return err
|
||||
}
|
||||
defer func() {
|
||||
err := dbCacheClient.Close(ctx)
|
||||
if err != nil {
|
||||
l.Errorf("closing db: %s", err.Error())
|
||||
}
|
||||
}()
|
||||
|
||||
// create http client
|
||||
httpClient, err := http.NewClient(
|
||||
ctx,
|
||||
fmt.Sprintf("Go-http-client/2.0 (%s/%s; +https://%s/)",
|
||||
viper.GetString(config.Keys.ApplicationName),
|
||||
viper.GetString(config.Keys.SoftwareVersion),
|
||||
viper.GetString(config.Keys.ServerExternalHostname),
|
||||
),
|
||||
viper.GetInt(config.Keys.HTTPClientTimeout),
|
||||
)
|
||||
if err != nil {
|
||||
l.Errorf("http client: %s", err.Error())
|
||||
cancel()
|
||||
|
||||
return err
|
||||
}
|
||||
|
||||
// create kv client
|
||||
kvClient, err := redis.New(ctx)
|
||||
if err != nil {
|
||||
l.Errorf("redis: %s", err.Error())
|
||||
cancel()
|
||||
|
||||
return err
|
||||
}
|
||||
defer func() {
|
||||
err := kvClient.Close(ctx)
|
||||
if err != nil {
|
||||
l.Errorf("closing redis: %s", err.Error())
|
||||
}
|
||||
}()
|
||||
|
||||
// create tokenizer
|
||||
tokz, err := token.New()
|
||||
if err != nil {
|
||||
l.Errorf("create tokenizer: %s", err.Error())
|
||||
cancel()
|
||||
|
||||
return err
|
||||
}
|
||||
|
||||
// create logic module
|
||||
l.Debug("creating logic module")
|
||||
logicMod, err := logic1.New(ctx, clockMod, dbCacheClient, httpClient, kvClient, tokz)
|
||||
if err != nil {
|
||||
l.Errorf("logic: %s", err.Error())
|
||||
cancel()
|
||||
|
||||
return err
|
||||
}
|
||||
|
||||
// create runner
|
||||
// create runner
|
||||
l.Debug("creating runner module")
|
||||
runnerAddr := viper.GetString(config.Keys.RedisAddress)
|
||||
runnerPassword := viper.GetString(config.Keys.RedisPassword)
|
||||
runnerDB := viper.GetInt(config.Keys.RedisDB)
|
||||
|
||||
if viper.GetString(config.Keys.RunnerAddress) != "" {
|
||||
runnerAddr = viper.GetString(config.Keys.RunnerAddress)
|
||||
runnerPassword = viper.GetString(config.Keys.RunnerPassword)
|
||||
runnerDB = viper.GetInt(config.Keys.RunnerDB)
|
||||
}
|
||||
|
||||
runnerMod, err := asynq.New(&asynq.Config{
|
||||
Logic: logicMod,
|
||||
|
||||
Concurrency: viper.GetInt(config.Keys.RunnerConcurrency),
|
||||
Address: runnerAddr,
|
||||
Password: runnerPassword,
|
||||
DB: runnerDB,
|
||||
})
|
||||
if err != nil {
|
||||
l.Errorf("runner server: %s", err.Error())
|
||||
cancel()
|
||||
|
||||
return err
|
||||
}
|
||||
logicMod.SetRunner(runnerMod)
|
||||
|
||||
// scheduler
|
||||
scheduler, err := runnerMod.NewScheduler()
|
||||
if err != nil {
|
||||
l.Errorf("getting scheduler: %s", err.Error())
|
||||
cancel()
|
||||
|
||||
return err
|
||||
}
|
||||
err = scheduler.Start()
|
||||
if err != nil {
|
||||
l.Errorf("starting scheduler: %s", err.Error())
|
||||
cancel()
|
||||
|
||||
return err
|
||||
}
|
||||
defer func() {
|
||||
l.Debug("closing scheduler")
|
||||
err := scheduler.Stop()
|
||||
if err != nil {
|
||||
l.Errorf("closing scheduler: %s", err.Error())
|
||||
}
|
||||
}()
|
||||
|
||||
// ** start application **
|
||||
errChan := make(chan error)
|
||||
|
||||
// Wait for SIGINT and SIGTERM (HIT CTRL-C)
|
||||
stopSigChan := make(chan os.Signal)
|
||||
signal.Notify(stopSigChan, syscall.SIGINT, syscall.SIGTERM)
|
||||
|
||||
// start metrics server
|
||||
go func(m *metrics.Module, errChan chan error) {
|
||||
l.Debug("starting metrics server")
|
||||
err := m.Start()
|
||||
if err != nil {
|
||||
errChan <- fmt.Errorf("metrics server: %s", err.Error())
|
||||
}
|
||||
}(metricsServer, errChan)
|
||||
|
||||
// wait for event
|
||||
select {
|
||||
case sig := <-stopSigChan:
|
||||
l.Infof("got sig: %s", sig)
|
||||
cancel()
|
||||
case err := <-errChan:
|
||||
l.Fatal(err.Error())
|
||||
cancel()
|
||||
}
|
||||
|
||||
<-ctx.Done()
|
||||
l.Infof("done")
|
||||
return nil
|
||||
}
|
||||
@ -0,0 +1,15 @@
|
||||
package flag
|
||||
|
||||
import (
|
||||
"git.ptzo.gdn/feditools/relay/internal/config"
|
||||
"github.com/spf13/cobra"
|
||||
)
|
||||
|
||||
// Scheduler adds all flags for running the scheduler.
|
||||
func Scheduler(cmd *cobra.Command, values config.Values) {
|
||||
Redis(cmd, values)
|
||||
Runner(cmd, values)
|
||||
|
||||
// metrics
|
||||
cmd.PersistentFlags().String(config.Keys.MetricsHTTPBind, values.MetricsHTTPBind, usage.MetricsHTTPBind)
|
||||
}
|
||||
@ -0,0 +1,33 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"git.ptzo.gdn/feditools/relay/cmd/relay/action/scheduler"
|
||||
"git.ptzo.gdn/feditools/relay/cmd/relay/flag"
|
||||
"git.ptzo.gdn/feditools/relay/internal/config"
|
||||
"github.com/spf13/cobra"
|
||||
)
|
||||
|
||||
// schedulerCommands returns the 'scheduler' subcommand
|
||||
func schedulerCommands() *cobra.Command {
|
||||
schedulerCmd := &cobra.Command{
|
||||
Use: "scheduler",
|
||||
Short: "controls a scheduler",
|
||||
}
|
||||
|
||||
schedulerStartCmd := &cobra.Command{
|
||||
Use: "start",
|
||||
Short: "start the relay scheduler",
|
||||
PreRunE: func(cmd *cobra.Command, args []string) error {
|
||||
return preRun(cmd)
|
||||
},
|
||||
RunE: func(cmd *cobra.Command, args []string) error {
|
||||
return run(cmd.Context(), scheduler.Start)
|
||||
},
|
||||
}
|
||||
|
||||
flag.Scheduler(schedulerStartCmd, config.Defaults)
|
||||
|
||||
schedulerCmd.AddCommand(schedulerStartCmd)
|
||||
|
||||
return schedulerCmd
|
||||
}
|
||||
@ -0,0 +1,61 @@
|
||||
package asynq
|
||||
|
||||
import (
|
||||
"git.ptzo.gdn/feditools/relay/internal/runner"
|
||||
"github.com/hibiken/asynq"
|
||||
)
|
||||
|
||||
func (r *Runner) NewScheduler() (*Scheduler, error) {
|
||||
l := logger.WithField("func", "NewScheduler")
|
||||
|
||||
s := asynq.NewScheduler(
|
||||
asynq.RedisClientOpt{
|
||||
Addr: r.address,
|
||||
Password: r.password,
|
||||
DB: r.db,
|
||||
},
|
||||
nil,
|
||||
)
|
||||
|
||||
// delivery error
|
||||
taskMaintDeliveryErrorTimeout := asynq.NewTask(TypeMaintDeliveryErrorTimeout, nil)
|
||||
entryID, err := s.Register("0 1 * * *", taskMaintDeliveryErrorTimeout, asynq.Queue(runner.QueuePriority))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
l.Debugf("registered task %s: %q\n", TypeMaintDeliveryErrorTimeout, entryID)
|
||||
|
||||
// update accounts
|
||||
taskUpdateAccountsInfo := asynq.NewTask(TypeUpdateAccountsInfo, nil)
|
||||
entryID, err = s.Register("0 2 * * *", taskUpdateAccountsInfo, asynq.Queue(runner.QueuePriority))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
l.Debugf("registered task %s: %q\n", TypeUpdateAccountsInfo, entryID)
|
||||
|
||||
// update instances
|
||||
taskUpdateInstancesInfo := asynq.NewTask(TypeUpdateInstancesInfo, nil)
|
||||
entryID, err = s.Register("0 3 * * *", taskUpdateInstancesInfo, asynq.Queue(runner.QueuePriority))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
l.Debugf("registered task %s: %q\n", TypeUpdateInstancesInfo, entryID)
|
||||
|
||||
return &Scheduler{
|
||||
scheduler: s,
|
||||
}, nil
|
||||
}
|
||||
|
||||
type Scheduler struct {
|
||||
scheduler *asynq.Scheduler
|
||||
}
|
||||
|
||||
func (s *Scheduler) Start() error {
|
||||
return s.scheduler.Start()
|
||||
}
|
||||
|
||||
func (s *Scheduler) Stop() error {
|
||||
s.scheduler.Shutdown()
|
||||
|
||||
return nil
|
||||
}
|
||||
@ -0,0 +1,23 @@
|
||||
package asynq
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"github.com/hibiken/asynq"
|
||||
)
|
||||
|
||||
const TypeUpdateAccountsInfo = "update:accounts"
|
||||
|
||||
func (r *Runner) handleUpdateAccountsInfo(ctx context.Context, _ *asynq.Task) error {
|
||||
ctx, span := r.tracer.Start(ctx, "handleUpdateAccountsInfo")
|
||||
defer span.End()
|
||||
|
||||
//l := logger.WithField("func", "handleUpdateAccountsInfo")
|
||||
|
||||
// process activity
|
||||
if err := r.logic.EnqueueAccountInfoUpdates(ctx); err != nil {
|
||||
return fmt.Errorf("%s: %w", err.Error(), asynq.SkipRetry)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
@ -0,0 +1,23 @@
|
||||
package asynq
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"github.com/hibiken/asynq"
|
||||
)
|
||||
|
||||
const TypeUpdateInstancesInfo = "update:instances"
|
||||
|
||||
func (r *Runner) handleUpdateInstancesInfo(ctx context.Context, _ *asynq.Task) error {
|
||||
ctx, span := r.tracer.Start(ctx, "handleUpdateInstancesInfo")
|
||||
defer span.End()
|
||||
|
||||
//l := logger.WithField("func", "handleUpdateInstancesInfo")
|
||||
|
||||
// process activity
|
||||
if err := r.logic.EnqueueInstanceInfoUpdates(ctx); err != nil {
|
||||
return fmt.Errorf("%s: %w", err.Error(), asynq.SkipRetry)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
@ -1,43 +0,0 @@
|
||||
package faktory
|
||||
|
||||
import (
|
||||
"context"
|
||||
faktory "github.com/contribsys/faktory/client"
|
||||
worker "github.com/contribsys/faktory_worker_go"
|
||||
"github.com/sirupsen/logrus"
|
||||
"strconv"
|
||||
)
|
||||
|
||||
func (r *Runner) EnqueueUpdateAccountInfo(_ context.Context, accountID int64) error {
|
||||
retry := 0
|
||||
|
||||
job := faktory.NewJob(JobUpdateAccountInfo, strconv.FormatInt(accountID, 10))
|
||||
job.Retry = &retry
|
||||
|
||||
return r.manager.Pool.With(func(conn *faktory.Client) error {
|
||||
return conn.Push(job)
|
||||
})
|
||||
}
|
||||
|
||||
func (r *Runner) updateAccountInfo(ctx context.Context, args ...interface{}) error {
|
||||
help := worker.HelperFor(ctx)
|
||||
|
||||
l := logger.WithFields(logrus.Fields{
|
||||
"func": "updateAccountInfo",
|
||||
"jid": help.Jid(),
|
||||
})
|
||||
|
||||
if len(args) != 1 {
|
||||
l.Errorf("wrong number of arguments, got: %d, want: %d", len(args), 2)
|
||||
}
|
||||
|
||||
// cast arguments
|
||||
accountID, err := getInt64(args, 0)
|
||||
if err != nil {
|
||||
l.Error(err.Error())
|
||||
|
||||
return err
|
||||
}
|
||||
|
||||
return r.logic.UpdateAccountInfo(ctx, help.Jid(), accountID)
|
||||
}
|
||||
@ -1,120 +0,0 @@
|
||||
package faktory
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"git.ptzo.gdn/feditools/go-lib/fedihelper"
|
||||
"git.ptzo.gdn/feditools/relay/internal/runner"
|
||||
faktory "github.com/contribsys/faktory/client"
|
||||
worker "github.com/contribsys/faktory_worker_go"
|
||||
"github.com/sirupsen/logrus"
|
||||
"net/url"
|
||||
"strconv"
|
||||
)
|
||||
|
||||
func (r *Runner) EnqueueDeliverActivity(_ context.Context, instanceID int64, activity fedihelper.Activity) error {
|
||||
retry := 8
|
||||
|
||||
job := faktory.NewJob(JobDeliverActivity, strconv.FormatInt(instanceID, 10), activity)
|
||||
job.Queue = runner.QueueDelivery
|
||||
job.Retry = &retry
|
||||
|
||||
return r.manager.Pool.With(func(conn *faktory.Client) error {
|
||||
return conn.Push(job)
|
||||
})
|
||||
}
|
||||
|
||||
func (r *Runner) deliverActivity(ctx context.Context, args ...interface{}) error {
|
||||
help := worker.HelperFor(ctx)
|
||||
|
||||
l := logger.WithFields(logrus.Fields{
|
||||
"func": "deliverActivity",
|
||||
"jid": help.Jid(),
|
||||
})
|
||||
|
||||
if len(args) != 2 {
|
||||
l.Errorf("wrong number of arguments, got: %d, want: %d", len(args), 2)
|
||||
}
|
||||
|
||||
// cast arguments
|
||||
instanceIDStr, ok := args[0].(string)
|
||||
if !ok {
|
||||
l.Errorf("argument 0 is not an string")
|
||||
|
||||
return fmt.Errorf("argument 0 is not an int")
|
||||
}
|
||||
instanceID, err := strconv.ParseInt(instanceIDStr, 10, 64)
|
||||
if err != nil {
|
||||
l.Errorf("cant parse int from argument 0: %s", err.Error())
|
||||
|
||||
return fmt.Errorf("cant parse int from argument 0: %s", err.Error())
|
||||
|
||||
}
|
||||
activity, ok := args[1].(map[string]interface{})
|
||||
if !ok {
|
||||
l.Errorf("argument 1 is not an activity")
|
||||
|
||||
return fmt.Errorf("argument 1 is not an activity")
|
||||
}
|
||||
|
||||
return r.logic.DeliverActivity(ctx, help.Jid(), instanceID, activity, false)
|
||||
}
|
||||
|
||||
func (r *Runner) EnqueueInboxActivity(_ context.Context, instanceID int64, actorIRI string, activity fedihelper.Activity) error {
|
||||
job := faktory.NewJob(JobInboxActivity, strconv.FormatInt(instanceID, 10), actorIRI, activity)
|
||||
job.Queue = runner.QueueDefault
|
||||
|
||||
return r.manager.Pool.With(func(conn *faktory.Client) error {
|
||||
return conn.Push(job)
|
||||
})
|
||||
}
|
||||
|
||||
func (r *Runner) inboxActivity(ctx context.Context, args ...interface{}) error {
|
||||
help := worker.HelperFor(ctx)
|
||||
|
||||
l := logger.WithFields(logrus.Fields{
|
||||
"func": "inboxActivity",
|
||||
"jid": help.Jid(),
|
||||
})
|
||||
|
||||
if len(args) != 3 {
|
||||
l.Errorf("wrong number of arguments, got: %d, want: %d", len(args), 2)
|
||||
}
|
||||
|
||||
// cast arguments
|
||||
instanceIDStr, ok := args[0].(string)
|
||||
if !ok {
|
||||
l.Errorf("argument 0 is not an string, got %T", args[0])
|
||||
|
||||
return fmt.Errorf("argument 0 is not an int")
|
||||
}
|
||||
instanceID, err := strconv.ParseInt(instanceIDStr, 10, 64)
|
||||
if err != nil {
|
||||
l.Errorf("cant parse int from argument 0: %s", err.Error())
|
||||
|
||||
return fmt.Errorf("cant parse int from argument 0: %s", err.Error())
|
||||
}
|
||||
|
||||
actorID, ok := args[1].(string)
|
||||
if !ok {
|
||||
l.Errorf("argument 1 is not an string, got %T", args[1])
|
||||
|
||||
return fmt.Errorf("argument 1 is not an actor")
|
||||
}
|
||||
actorIRI, err := url.Parse(actorID)
|
||||
if err != nil {
|
||||
l.Errorf("cant parse url from argument 1: %s", err.Error())
|
||||
|
||||
return fmt.Errorf("cant parse url from argument 1: %s", err.Error())
|
||||
}
|
||||
|
||||
activity, ok := args[2].(map[string]interface{})
|
||||
if !ok {
|
||||
l.Errorf("argument 2 is not an activity, got %T", args[2])
|
||||
|
||||
return fmt.Errorf("argument 2 is not an activity")
|
||||
}
|
||||
|
||||
// process activity
|
||||
return r.logic.ProcessActivity(ctx, help.Jid(), instanceID, actorIRI, activity)
|
||||
}
|
||||
@ -1,102 +0,0 @@
|
||||
package faktory
|
||||
|
||||
import (
|
||||
"context"
|
||||
faktory "github.com/contribsys/faktory/client"
|
||||
worker "github.com/contribsys/faktory_worker_go"
|
||||
"github.com/sirupsen/logrus"
|
||||
"strconv"
|
||||
)
|
||||
|
||||
func (r *Runner) EnqueueProcessBlockAdd(_ context.Context, blockID int64) error {
|
||||
job := faktory.NewJob(JobProcessBlockAdd, strconv.FormatInt(blockID, 10))
|
||||
|
||||
return r.manager.Pool.With(func(conn *faktory.Client) error {
|
||||
return conn.Push(job)
|
||||
})
|
||||
}
|
||||
|
||||
func (r *Runner) EnqueueProcessBlockDelete(_ context.Context, blockID int64) error {
|
||||
job := faktory.NewJob(JobProcessBlockDelete, strconv.FormatInt(blockID, 10))
|
||||
|
||||
return r.manager.Pool.With(func(conn *faktory.Client) error {
|
||||
return conn.Push(job)
|
||||
})
|
||||
}
|
||||
|
||||
func (r *Runner) EnqueueProcessBlockUpdate(_ context.Context, blockID int64) error {
|
||||
job := faktory.NewJob(JobProcessBlockUpdate, strconv.FormatInt(blockID, 10))
|
||||
|
||||
return r.manager.Pool.With(func(conn *faktory.Client) error {
|
||||
return conn.Push(job)
|
||||
})
|
||||
}
|
||||
|
||||
func (r *Runner) processBlockAdd(ctx context.Context, args ...interface{}) error {
|
||||
help := worker.HelperFor(ctx)
|
||||
|
||||
l := logger.WithFields(logrus.Fields{
|
||||
"func": "processBlockAdd",
|
||||
"jid": help.Jid(),
|
||||
})
|
||||
|
||||
if len(args) != 1 {
|
||||
l.Errorf("wrong number of arguments, got: %d, want: %d", len(args), 2)
|
||||
}
|
||||
|
||||
// cast arguments
|
||||
blockID, err := getInt64(args, 0)
|
||||
if err != nil {
|
||||
l.Error(err.Error())
|
||||
|
||||
return err
|
||||
}
|
||||
|
||||
return r.logic.ProcessBlockAdd(ctx, blockID)
|
||||
}
|
||||
|
||||
func (r *Runner) processBlockDelete(ctx context.Context, args ...interface{}) error {
|
||||
help := worker.HelperFor(ctx)
|
||||
|
||||
l := logger.WithFields(logrus.Fields{
|
||||
"func": "processBlockDelete",
|
||||
"jid": help.Jid(),
|
||||
})
|
||||
|
||||
if len(args) != 1 {
|
||||
l.Errorf("wrong number of arguments, got: %d, want: %d", len(args), 2)
|
||||
}
|
||||
|
||||
// cast arguments
|
||||
blockID, err := getInt64(args, 0)
|
||||
if err != nil {
|
||||
l.Error(err.Error())
|
||||
|
||||
return err
|
||||
}
|
||||
|
||||
return r.logic.ProcessBlockDelete(ctx, blockID)
|
||||
}
|
||||
|
||||
func (r *Runner) processBlockUpdate(ctx context.Context, args ...interface{}) error {
|
||||
help := worker.HelperFor(ctx)
|
||||
|
||||
l := logger.WithFields(logrus.Fields{
|
||||
"func": "processBlockDelete",
|
||||
"jid": help.Jid(),
|
||||
})
|
||||
|
||||
if len(args) != 1 {
|
||||
l.Errorf("wrong number of arguments, got: %d, want: %d", len(args), 2)
|
||||
}
|
||||
|
||||
// cast arguments
|
||||
blockID, err := getInt64(args, 0)
|
||||
if err != nil {
|
||||
l.Error(err.Error())
|
||||
|
||||
return err
|
||||
}
|
||||
|
||||
return r.logic.ProcessBlockUpdate(ctx, blockID)
|
||||
}
|
||||
@ -1,15 +0,0 @@
|
||||
package faktory
|
||||
|
||||
const (
|
||||
twice = 2
|
||||
|
||||
JobDeliverActivity = "DeliverActivity"
|
||||
JobInboxActivity = "InboxActivity"
|
||||
JobMaintDeliveryErrorTimeout = "MaintDeliveryErrorTimeout"
|
||||
JobProcessBlockAdd = "ProcessBlockAdd"
|
||||
JobProcessBlockDelete = "ProcessBlockDelete"
|
||||
JobProcessBlockUpdate = "ProcessBlockUpdate"
|
||||
JobSendNotification = "SendNotification"
|
||||
JobUpdateAccountInfo = "UpdateAccountInfo"
|
||||
JobUpdateInstanceInfo = "UpdateInstanceInfo"
|
||||
)
|
||||
@ -1,43 +0,0 @@
|
||||
package faktory
|
||||
|
||||
import (
|
||||
"context"
|
||||
faktory "github.com/contribsys/faktory/client"
|
||||
worker "github.com/contribsys/faktory_worker_go"
|
||||
"github.com/sirupsen/logrus"
|
||||
"strconv"
|
||||
)
|
||||
|
||||
func (r *Runner) EnqueueUpdateInstanceInfo(_ context.Context, instanceID int64) error {
|
||||
retry := 0
|
||||
|
||||
job := faktory.NewJob(JobUpdateInstanceInfo, strconv.FormatInt(instanceID, 10))
|
||||
job.Retry = &retry
|
||||
|
||||
return r.manager.Pool.With(func(conn *faktory.Client) error {
|
||||
return conn.Push(job)
|
||||
})
|
||||
}
|
||||
|
||||
func (r *Runner) updateInstanceInfo(ctx context.Context, args ...interface{}) error {
|
||||
help := worker.HelperFor(ctx)
|
||||
|
||||
l := logger.WithFields(logrus.Fields{
|
||||
"func": "updateInstanceInfo",
|
||||
"jid": help.Jid(),
|
||||
})
|
||||
|
||||
if len(args) != 1 {
|
||||
l.Errorf("wrong number of arguments, got: %d, want: %d", len(args), 2)
|
||||
}
|
||||
|
||||
// cast arguments
|
||||
instanceID, err := getInt64(args, 0)
|
||||
if err != nil {
|
||||
l.Error(err.Error())
|
||||
|
||||
return err
|
||||
}
|
||||
|
||||
return r.logic.UpdateInstanceInfo(ctx, help.Jid(), instanceID)
|
||||
}
|
||||
@ -1,9 +0,0 @@
|
||||
package faktory
|
||||
|
||||
import (
|
||||
"git.ptzo.gdn/feditools/relay/internal/log"
|
||||
)
|
||||
|
||||
type empty struct{}
|
||||
|
||||
var logger = log.WithPackageField(empty{})
|
||||
@ -1,32 +0,0 @@
|
||||
package faktory
|
||||
|
||||
import (
|
||||
"context"
|
||||
faktory "github.com/contribsys/faktory/client"
|
||||
worker "github.com/contribsys/faktory_worker_go"
|
||||
"github.com/sirupsen/logrus"
|
||||
)
|
||||
|
||||
func (r *Runner) EnqueueMaintDeliveryErrorTimeout(_ context.Context) error {
|
||||
job := faktory.NewJob(JobMaintDeliveryErrorTimeout)
|
||||
job.Args = []interface{}{}
|
||||
|
||||
return r.manager.Pool.With(func(conn *faktory.Client) error {
|
||||
return conn.Push(job)
|
||||
})
|
||||
}
|
||||
|
||||
func (r *Runner) maintDeliveryErrorTimeout(ctx context.Context, args ...interface{}) error {
|
||||
help := worker.HelperFor(ctx)
|
||||
|
||||
l := logger.WithFields(logrus.Fields{
|
||||
"func": "maintDeliveryErrorTimeout",
|
||||
"jid": help.Jid(),
|
||||
})
|
||||
|
||||
if len(args) != 0 {
|
||||
l.Errorf("wrong number of arguments, got: %d, want: %d", len(args), 0)
|
||||
}
|
||||
|
||||
return r.logic.MaintDeliveryErrorTimeout(ctx, help.Jid())
|
||||
}
|
||||
@ -1,19 +0,0 @@
|
||||
package faktory
|
||||
|
||||
import (
|
||||
"context"
|
||||
"git.ptzo.gdn/feditools/relay/internal/runner"
|
||||
faktory "github.com/contribsys/faktory/client"
|
||||
"go.opentelemetry.io/otel/trace"
|
||||
)
|
||||
|
||||
func (r *Runner) Middleware(ctx context.Context, job *faktory.Job, next func(ctx context.Context) error) error {
|
||||
opts := []trace.SpanStartOption{
|
||||
trace.WithAttributes(runner.JobIDKey.String(job.Jid)),
|
||||
trace.WithSpanKind(trace.SpanKindConsumer),
|
||||
}
|
||||
tctx, main := r.tracer.Start(ctx, job.Type, opts...)
|
||||
defer main.End()
|
||||
|
||||
return next(tctx)
|
||||
}
|
||||
@ -1,49 +0,0 @@
|
||||
package faktory
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"git.ptzo.gdn/feditools/relay/internal/models"
|
||||
"git.ptzo.gdn/feditools/relay/internal/runner"
|
||||
faktory "github.com/contribsys/faktory/client"
|
||||
worker "github.com/contribsys/faktory_worker_go"
|
||||
"github.com/sirupsen/logrus"
|
||||
)
|
||||
|
||||
func (r *Runner) EnqueueSendNotification(_ context.Context, event models.EventType, metadata map[string]interface{}) error {
|
||||
job := faktory.NewJob(JobSendNotification, event, metadata)
|
||||
job.Queue = runner.QueuePriority
|
||||
|
||||
return r.manager.Pool.With(func(conn *faktory.Client) error {
|
||||
return conn.Push(job)
|
||||
})
|
||||
}
|
||||
|
||||
func (r *Runner) sendNotification(ctx context.Context, args ...interface{}) error {
|
||||
help := worker.HelperFor(ctx)
|
||||
|
||||