smutok/internal/freshrss/worker.go(view raw)
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 |
package freshrss
import (
"context"
"log/slog"
"net"
"net/url"
"sync"
"time"
"olexsmir.xyz/smutok/internal/store"
)
type Worker struct {
api *Client
store *store.Sqlite
apiServer string
writeToken string
}
func NewWorker(api *Client, store *store.Sqlite, writeToken, apiServer string) *Worker {
return &Worker{
api: api,
store: store,
writeToken: writeToken,
apiServer: apiServer,
}
}
func (w *Worker) Run(ctx context.Context) {
ticker := time.NewTicker(5 * time.Second)
defer ticker.Stop()
var wg sync.WaitGroup
for {
select {
case <-ctx.Done():
return
default:
if !w.isNetworkAvailable() {
slog.Info("worker: no internet connection")
continue
}
wg.Go(func() {
if err := w.pendingReads(ctx); err != nil {
slog.Error("worker: read", "err", err)
}
})
wg.Go(func() {
if err := w.pendingUnreads(ctx); err != nil {
slog.Error("worker: unread", "err", err)
}
})
wg.Go(func() {
if err := w.pendingStar(ctx); err != nil {
slog.Error("worker: star", "err", err)
}
})
wg.Go(func() {
if err := w.pendingUnstar(ctx); err != nil {
slog.Error("worker: unread", "err", err)
}
})
wg.Wait()
<-ticker.C
}
}
}
func (w *Worker) isNetworkAvailable() bool {
u, err := url.Parse(w.apiServer)
if err != nil {
slog.Error("invalid api url", "url", w.apiServer, "err", err)
return false
}
_, err = net.LookupHost(u.Host)
return err == nil
}
func (w *Worker) pendingReads(ctx context.Context) error {
slog.Debug("worker: pending read")
return w.handle(ctx, store.Read, StateRead, "")
}
func (w *Worker) pendingUnreads(ctx context.Context) error {
slog.Debug("worker: pending unread")
return w.handle(ctx, store.Unread, StateKeptUnread, StateRead)
}
func (w *Worker) pendingStar(ctx context.Context) error {
slog.Debug("worker: pending star")
return w.handle(ctx, store.Star, StateStarred, "")
}
func (w *Worker) pendingUnstar(ctx context.Context) error {
slog.Debug("worker: pending unstar")
return w.handle(ctx, store.Unstar, "", StateStarred)
}
func (w *Worker) handle(ctx context.Context, action store.Action, addState, rmState string) error {
articleIDs, err := w.store.GetPendingActions(ctx, action)
if err != nil {
return err
}
if len(articleIDs) == 0 {
slog.Debug("worker: no pending actions", "action", action)
return nil
}
if err := w.api.EditTag(ctx, w.writeToken, EditTag{
ItemID: articleIDs,
TagToAdd: addState,
TagToRemove: rmState,
}); err != nil {
return err
}
return w.store.DeletePendingActions(ctx, action, articleIDs)
}
|