use nervana to label posts

cmd/photocopy/main.go

···
 			&cli.BoolFlag{
 				Name: "with-backfill",
 			},
+			&cli.StringFlag{
+				Name: "nervana-endpoint",
+			},
+			&cli.StringFlag{
+				Name: "nervana-api-key",
+			},
 		},
 		Commands: cli.Commands{
 			&cli.Command{
···
 		ClickhouseUser:       cmd.String("clickhouse-user"),
 		ClickhousePass:       cmd.String("clickhouse-pass"),
 		RatelimitBypassKey:   cmd.String("ratelimit-bypass-key"),
+		NervanaEndpoint:      cmd.String("nervana-endpoint"),
+		NervanaApiKey:        cmd.String("nervana-api-key"),
 	})
 	if err != nil {
 		panic(err)

handle_create.go

···
 		IndexedAt: indexedAt,
 		Did:       did,
 		Lang:      lang,
+		Text:      rec.Text,
 	}
 
 	if rec.Reply != nil {
···
 
 	if err := p.inserters.postsInserter.Insert(ctx, post); err != nil {
 		return err
+	}
+
+	if rec.Text != "" {
+		go func(ctx context.Context, rec bsky.FeedPost, did, rkey string) {
+			ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
+			defer cancel()
+
+			nervanaItems, err := p.makeNervanaRequest(ctx, rec.Text)
+			if err != nil {
+				p.logger.Error("error making nervana items request", "error", err)
+				return
+			}
+
+			for _, ni := range nervanaItems {
+				postLabel := models.PostLabel{
+					Did:         did,
+					Rkey:        rkey,
+					Text:        ni.Text,
+					Label:       ni.Label,
+					EntityId:    ni.EntityId,
+					Description: ni.Description,
+					Topic:       "",
+				}
+				p.inserters.labelsInserter.Insert(ctx, postLabel)
+			}
+		}(ctx, rec, did, rkey)
 	}
 
 	return nil

models/post.go

···
 	QuoteUri  string    `ch:"quote_uri"`
 	QuoteDid  string    `ch:"quote_did"`
 	Lang      string    `ch:"lang"`
+	Text      string    `ch:"text"`
 }

models/post_label.go

···
+package models
+
+import "time"
+
+type PostLabel struct {
+	Did         string    `ch:"did"`
+	Rkey        string    `ch:"rkey"`
+	CreatedAt   time.Time `ch:"created_at"`
+	Text        string    `ch:"text"`
+	Label       string    `ch:"label"`
+	EntityId    string    `ch:"entity_id"`
+	Description string    `ch:"description"`
+	Topic       string    `ch:"topic"`
+}

nervana.go

···
+package photocopy
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+)
+
+type NervanaItem struct {
+	Text        string `json:"text"`
+	Label       string `json:"label"`
+	EntityId    string `json:"entityId"`
+	Description string `json:"description"`
+}
+
+func (p *Photocopy) newNervanaRequest(ctx context.Context, text string) (*http.Request, error) {
+	payload := map[string]string{
+		"text":     text,
+		"language": "en",
+	}
+
+	b, err := json.Marshal(payload)
+	if err != nil {
+		return nil, err
+	}
+
+	req, err := http.NewRequestWithContext(ctx, "GET", p.nervanaEndpoint, bytes.NewReader(b))
+
+	req.Header.Set("Authorization", "Bearer "+p.nervanaApiKey)
+
+	return req, err
+}
+
+func (p *Photocopy) makeNervanaRequest(ctx context.Context, text string) ([]NervanaItem, error) {
+	req, err := p.newNervanaRequest(ctx, text)
+	if err != nil {
+		return nil, err
+	}
+
+	resp, err := p.nervanaClient.Do(req)
+	if err != nil {
+		return nil, err
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode != 200 {
+		io.Copy(io.Discard, resp.Body)
+		return nil, fmt.Errorf("received non-200 response code: %d", resp.StatusCode)
+	}
+
+	var nervanaResp []NervanaItem
+	if err := json.NewDecoder(resp.Body).Decode(&nervanaResp); err != nil {
+		return nil, err
+	}
+
+	return nervanaResp, nil
+}

photocopy.go

···
 	ratelimitBypassKey string
 
 	conn driver.Conn
+
+	nervanaClient   *http.Client
+	nervanaEndpoint string
+	nervanaApiKey   string
 }
 
 type Inserters struct {
···
 	plcInserter          *clickhouse_inserter.Inserter
 	recordsInserter      *clickhouse_inserter.Inserter
 	deletesInserter      *clickhouse_inserter.Inserter
+	labelsInserter       *clickhouse_inserter.Inserter
 }
 
 type Args struct {
···
 	ClickhouseUser       string
 	ClickhousePass       string
 	RatelimitBypassKey   string
+	NervanaEndpoint      string
+	NervanaApiKey        string
 }
 
 func New(ctx context.Context, args *Args) (*Photocopy, error) {
···
 		return nil, err
 	}
 
+	li, err := clickhouse_inserter.New(ctx, &clickhouse_inserter.Args{
+		PrometheusCounterPrefix: "photocopy_labels",
+		Histogram:               insertionsHist,
+		BatchSize:               100,
+		Logger:                  p.logger,
+		Conn:                    conn,
+		Query:                   "INSERT INTO post_label (did, rkey, text, label, entity_id, description, topic, created_at)",
+		RateLimit:               3,
+	})
+	if err != nil {
+		return nil, err
+	}
+
 	is := &Inserters{
 		followsInserter:      fi,
 		postsInserter:        pi,
 		interactionsInserter: ii,
 		recordsInserter:      ri,
 		deletesInserter:      di,
+		labelsInserter:       li,
 	}
 
 	p.inserters = is
···
 
 	p.inserters.plcInserter = plci
 	p.plcScraper = plcs
+
+	if args.NervanaApiKey != "" && args.NervanaEndpoint != "" {
+		p.nervanaClient = &http.Client{
+			Timeout: 5 * time.Second,
+		}
+		p.nervanaEndpoint = args.NervanaEndpoint
+		p.nervanaApiKey = args.NervanaApiKey
+	}
 
 	return p, nil
 }