feat: consume interests events from queues

awakari · Aug 2, 2024 · 6851a89 · 6851a89
1 parent 9cadb53
commit 6851a89
Show file tree

Hide file tree

Showing 5 changed files with 77 additions and 56 deletions.
diff --git a/config/config.go b/config/config.go
@@ -36,7 +36,7 @@ type MastodonConfig struct {
 		UserAgent string `envconfig:"API_MASTODON_USER_AGENT" default:"awakari" required:"true""`
 	}
 	Endpoint struct {
-		Accounts string `envconfig:"API_MASTODON_ENDPOINT_SEARCH" default:"https://mastodon.social/api/v1/accounts" required:"true"`
+		Accounts string `envconfig:"API_MASTODON_ENDPOINT_ACCOUNTS" default:"https://mastodon.social/api/v1/accounts" required:"true"`
 		Search   string `envconfig:"API_MASTODON_ENDPOINT_SEARCH" default:"https://mastodon.social/api/v2/search" required:"true"`
 		Stream   string `envconfig:"API_MASTODON_ENDPOINT_STREAM" default:"https://streaming.mastodon.social/api/v1/streaming/public?remote=false&only_media=false" required:"true"`
 	}
@@ -51,7 +51,8 @@ type MastodonConfig struct {
 }
 
 type QueueConfig struct {
-	Uri              string `envconfig:"API_QUEUE_URI" default:"queue:50051" required:"true"`
+	BackoffError     time.Duration `envconfig:"API_QUEUE_BACKOFF_ERROR" default:"1s" required:"true"`
+	Uri              string        `envconfig:"API_QUEUE_URI" default:"queue:50051" required:"true"`
 	InterestsCreated struct {
 		BatchSize uint32 `envconfig:"API_QUEUE_INTERESTS_CREATED_BATCH_SIZE" default:"1" required:"true"`
 		Name      string `envconfig:"API_QUEUE_INTERESTS_CREATED_NAME" default:"source-search" required:"true"`

diff --git a/helm/int-mastodon/templates/deployment.yaml b/helm/int-mastodon/templates/deployment.yaml
@@ -46,6 +46,8 @@ spec:
               value: "{{ .Values.api.event.type }}"
             - name: LOG_LEVEL
               value: "{{ .Values.log.level }}"
+            - name: API_MASTODON_ENDPOINT_ACCOUNTS
+              value: "{{ .Values.mastodon.endpoint.accounts }}"
             - name: API_MASTODON_ENDPOINT_SEARCH
               value: "{{ .Values.mastodon.endpoint.search }}"
             - name: API_MASTODON_ENDPOINT_STREAM

diff --git a/helm/int-mastodon/values.yaml b/helm/int-mastodon/values.yaml
@@ -73,7 +73,16 @@ nodeSelector: {}
 
 tolerations: []
 
-affinity: {}
+affinity:
+  nodeAffinity:
+    preferredDuringSchedulingIgnoredDuringExecution:
+      - weight: 1
+        preference:
+          matchExpressions:
+            - key: spot
+              operator: In
+              values:
+                - "true"
 
 api:
   activitypub:

diff --git a/main.go b/main.go
@@ -150,10 +150,13 @@ func consumeQueue(
 	log *slog.Logger,
 ) (err error) {
 	for {
-		_ = svcQueue.ReceiveMessages(ctx, name, subj, batchSize, func(evts []*pb.CloudEvent) (err error) {
+		err = svcQueue.ReceiveMessages(ctx, name, subj, batchSize, func(evts []*pb.CloudEvent) (err error) {
 			consumeEvents(ctx, svc, evts, typ, cfg, log)
 			return
 		})
+		if err != nil {
+			panic(err)
+		}
 	}
 }
 
@@ -165,12 +168,10 @@ func consumeEvents(
 	cfg config.Config,
 	log *slog.Logger,
 ) {
+	log.Debug(fmt.Sprintf("consumeEvents(%d, typ=%d))\n", len(evts), typ))
 	for _, evt := range evts {
+
 		interestId := evt.GetTextData()
-		var queries []string
-		if queriesComplAttr, queriesComplPresent := evt.Attributes[ceKeyQueriesCompl]; queriesComplPresent {
-			queries = strings.Split(queriesComplAttr.GetCeString(), "\n")
-		}
 		var groupId string
 		if groupIdAttr, groupIdIdPresent := evt.Attributes[ceKeyGroupId]; groupIdIdPresent {
 			groupId = groupIdAttr.GetCeString()
@@ -179,15 +180,27 @@ func consumeEvents(
 			log.Error(fmt.Sprintf("interest %s event type %d: empty group id, skipping", interestId, typ))
 			continue
 		}
-		if len(queries) > 0 {
-			for _, q := range queries {
-				_, _ = svc.SearchAndAdd(ctx, interestId, groupId, q, cfg.Api.Mastodon.Search.Limit, model.SearchTypeStatuses)
-			}
-		}
+
 		publicAttr, publicAttrPresent := evt.Attributes[ceKeyPublic]
-		if publicAttrPresent && publicAttr.GetCeBoolean() && typ == evtTypeInterestsCreated {
+		switch publicAttrPresent && publicAttr.GetCeBoolean() && typ == evtTypeInterestsCreated {
+		case true:
 			actor := interestId + "@" + cfg.Api.ActivityPub.Host
 			_, _ = svc.SearchAndAdd(ctx, interestId, groupId, actor, 1, model.SearchTypeAccounts)
+		default:
+			log.Debug(fmt.Sprintf("interest %s event type %d: public: %t/%t", interestId, typ, publicAttrPresent, publicAttr.GetCeBoolean()))
+		}
+
+		var queries []string
+		if queriesComplAttr, queriesComplPresent := evt.Attributes[ceKeyQueriesCompl]; queriesComplPresent {
+			queries = strings.Split(queriesComplAttr.GetCeString(), "\n")
+		}
+		switch len(queries) {
+		case 0:
+			log.Debug(fmt.Sprintf("interest %s event type %d: no queries, skipping the sources discovery", interestId, typ))
+		default:
+			for _, q := range queries {
+				_, _ = svc.SearchAndAdd(ctx, interestId, groupId, q, cfg.Api.Mastodon.Search.Limit, model.SearchTypeStatuses)
+			}
 		}
 	}
 	return

diff --git a/service/service.go b/service/service.go
@@ -58,11 +58,11 @@ func NewService(
 	}
 }
 
-func (m mastodon) SearchAndAdd(ctx context.Context, interestId, groupId, q string, limit uint32, typ model.SearchType) (n uint32, err error) {
-	var offset int
+func (m mastodon) SearchAndAdd(ctx context.Context, interestId, groupId, q string, limit uint32, typ model.SearchType) (n uint32, errs error) {
 	for n < limit {
-		reqQuery := "?q=" + url.QueryEscape(q) + "&type=" + typ.String() + "&resolve=true&offset=" + strconv.Itoa(offset) + "&limit=" + strconv.Itoa(int(limit))
+		reqQuery := "?q=" + url.QueryEscape(q) + "&type=" + typ.String() + "&resolve=true&offset=" + strconv.Itoa(int(n)) + "&limit=" + strconv.Itoa(int(limit-n))
 		var req *http.Request
+		var err error
 		req, err = http.NewRequestWithContext(ctx, http.MethodGet, m.cfg.Endpoint.Search+reqQuery, nil)
 		var resp *http.Response
 		if err == nil {
@@ -72,51 +72,40 @@ func (m mastodon) SearchAndAdd(ctx context.Context, interestId, groupId, q strin
 			resp, err = m.clientHttp.Do(req)
 		}
 		var data []byte
-		if err == nil {
+		if err == nil && resp != nil {
 			data, err = io.ReadAll(io.LimitReader(resp.Body, limitRespBodyLen))
 			_ = resp.Body.Close()
 		}
 		var results model.Results
 		if err == nil {
 			err = json.Unmarshal(data, &results)
 		}
-		if err == nil {
-			switch typ {
-			case model.SearchTypeStatuses:
-				countResults := len(results.Statuses)
-				if countResults == 0 {
-					break
-				}
-				offset += countResults
-				for _, st := range results.Statuses {
-					errSt := m.processFoundStatus(ctx, st, interestId, groupId, q)
-					switch errSt {
-					case nil:
-						n++
-					default:
-						err = errors.Join(err, errSt)
-					}
-					if n > limit {
-						break
-					}
-				}
-			case model.SearchTypeAccounts:
-				countResults := len(results.Accounts)
-				if countResults == 0 {
-					break
+		if err != nil {
+			errs = errors.Join(errs, err)
+			break
+		}
+		if typ == model.SearchTypeStatuses {
+			countResults := len(results.Statuses)
+			if countResults == 0 {
+				break
+			}
+			n += uint32(countResults)
+			for _, st := range results.Statuses {
+				err = m.processFoundStatus(ctx, st, interestId, groupId, q)
+				if err != nil {
+					err = errors.Join(errs, err)
 				}
-				offset += countResults
-				for _, acc := range results.Accounts {
-					errSt := m.processFoundAccount(ctx, acc, interestId, groupId, q, false)
-					switch errSt {
-					case nil:
-						n++
-					default:
-						err = errors.Join(err, errSt)
-					}
-					if n > limit {
-						break
-					}
+			}
+		} else if typ == model.SearchTypeAccounts {
+			countResults := len(results.Accounts)
+			if countResults == 0 {
+				break
+			}
+			n += uint32(countResults)
+			for _, acc := range results.Accounts {
+				err = m.processFoundAccount(ctx, acc, interestId, groupId, q, false)
+				if err != nil {
+					errs = errors.Join(errs, err)
 				}
 			}
 		}
@@ -170,7 +159,7 @@ func (m mastodon) processFoundAccount(ctx context.Context, acc model.Account, in
 
 func (m mastodon) follow(ctx context.Context, acc model.Account) (err error) {
 	var req *http.Request
-	req, err = http.NewRequestWithContext(ctx, http.MethodPost, m.cfg.Endpoint.Search+"/"+acc.Id+"/follow", strings.NewReader(`{"reblogs":true}`))
+	req, err = http.NewRequestWithContext(ctx, http.MethodPost, m.cfg.Endpoint.Accounts+"/"+acc.Id+"/follow", nil)
 	var resp *http.Response
 	if err == nil {
 		req.Header.Add("Accept", "application/json")
@@ -182,7 +171,14 @@ func (m mastodon) follow(ctx context.Context, acc model.Account) (err error) {
 		defer resp.Body.Close()
 		if resp.StatusCode != http.StatusOK {
 			data, _ := io.ReadAll(io.LimitReader(resp.Body, limitRespBodyLenErr))
-			err = fmt.Errorf("failed to follow the account %s: %s", acc.Acct, string(data))
+			err = fmt.Errorf(
+				"failed to follow the account %s: request_url=%s, request_headers=%+v, response=%d/%s",
+				acc.Acct,
+				req.URL,
+				req.Header,
+				resp.StatusCode,
+				string(data),
+			)
 		}
 	}
 	return