Make API publish new index event

internal/api/api.go

@@ -223,6 +223,21 @@ func (state *State) addResource(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
+	// Publish linked event
+	if err := state.pub.PublishEvent(&event.NewIndexEvent{
+		URL:         res.URL,
+		Body:        res.Body,
+		Time:        res.Time,
+		Title:       res.Title,
+		Meta:        res.Meta,
+		Description: res.Description,
+		Headers:     res.Headers,
+	}); err != nil {
+		log.Err(err).Msg("Error while publishing new index event")
+		w.WriteHeader(http.StatusInternalServerError)
+		return
+	}
+
 	log.Info().Str("url", res.URL).Msg("Successfully saved resource")
 
 	writeJSON(w, res)

internal/api/api_test.go

@@ -160,6 +160,7 @@ func TestAddResource(t *testing.T) {
 
 	dbMock := database_mock.NewMockDatabase(mockCtrl)
 	configClientMock := client_mock.NewMockClient(mockCtrl)
+	pubMock := event_mock.NewMockPublisher(mockCtrl)
 
 	dbMock.EXPECT().CountResources(&searchParamsMatcher{target: api.ResSearchParams{
 		URL:        "https://example.onion",
@@ -177,10 +178,20 @@ func TestAddResource(t *testing.T) {
 		Headers:     map[string]string{"Content-Type": "application/html", "Server": "Traefik"},
 	})
 
+	pubMock.EXPECT().PublishEvent(&event.NewIndexEvent{
+		URL:         "https://example.onion",
+		Body:        "TheBody",
+		Title:       "Example",
+		Time:        time.Time{},
+		Meta:        map[string]string{"content": "content-meta"},
+		Description: "the description",
+		Headers:     map[string]string{"Content-Type": "application/html", "Server": "Traefik"},
+	})
+
 	configClientMock.EXPECT().GetRefreshDelay().Return(client.RefreshDelay{Delay: 5 * time.Hour}, nil)
 	configClientMock.EXPECT().GetForbiddenHostnames().Return([]client.ForbiddenHostname{}, nil)
 
-	s := State{db: dbMock, configClient: configClientMock}
+	s := State{db: dbMock, configClient: configClientMock, pub: pubMock}
 
 	s.addResource(rec, req)
 	if rec.Code != http.StatusOK {

internal/archiver/archiver.go

@@ -51,7 +51,7 @@ func (state *State) Initialize(provider process.Provider) error {
 // Subscribers return the process subscribers
 func (state *State) Subscribers() []process.SubscriberDef {
 	return []process.SubscriberDef{
-		{Exchange: event.NewResourceExchange, Queue: "archivingQueue", Handler: state.handleNewResourceEvent},
+		{Exchange: event.NewIndexExchange, Queue: "archivingQueue", Handler: state.handleNewResourceEvent},
 	}
 }
 
@@ -61,7 +61,7 @@ func (state *State) HTTPHandler(provider process.Provider) http.Handler {
 }
 
 func (state *State) handleNewResourceEvent(subscriber event.Subscriber, msg event.RawMessage) error {
-	var evt event.NewResourceEvent
+	var evt event.NewIndexEvent
 	if err := subscriber.Read(&msg, &evt); err != nil {
 		return err
 	}
@@ -80,7 +80,7 @@ func (state *State) handleNewResourceEvent(subscriber event.Subscriber, msg even
 	return nil
 }
 
-func formatResource(evt *event.NewResourceEvent) ([]byte, error) {
+func formatResource(evt *event.NewIndexEvent) ([]byte, error) {
 	builder := strings.Builder{}
 
 	// First headers

internal/archiver/archiver_test.go

@@ -20,8 +20,8 @@ func TestHandleNewResourceEvent(t *testing.T) {
 
 	msg := event.RawMessage{}
 	subscriberMock.EXPECT().
-		Read(&msg, &event.NewResourceEvent{}).
-		SetArg(1, event.NewResourceEvent{
+		Read(&msg, &event.NewIndexEvent{}).
+		SetArg(1, event.NewIndexEvent{
 			URL:     "https://example.onion",
 			Body:    "Hello, world",
 			Headers: map[string]string{"Server": "Traefik", "Content-Type": "application/html"},
@@ -37,7 +37,7 @@ func TestHandleNewResourceEvent(t *testing.T) {
 }
 
 func TestFormatResource(t *testing.T) {
-	evt := &event.NewResourceEvent{
+	evt := &event.NewIndexEvent{
 		URL:     "https://google.com",
 		Body:    "Hello, world",
 		Headers: map[string]string{"Server": "Traefik", "Content-Type": "text/html"},

internal/event/event.go

@@ -13,6 +13,8 @@ const (
 	TimeoutURLExchange = "url.timeout"
 	// NewResourceExchange is the exchange used when a new resource has been crawled
 	NewResourceExchange = "resource.new"
+	// NewIndexExchange is the exchange used when a resource has been indexed
+	NewIndexExchange = "index.new"
 	// ConfigExchange is the exchange used to dispatch new configuration
 	ConfigExchange = "config"
 )
@@ -65,3 +67,19 @@ type NewResourceEvent struct {
 func (msg *NewResourceEvent) Exchange() string {
 	return NewResourceExchange
 }
+
+// NewResourceEvent represent a indexed resource
+type NewIndexEvent struct {
+	URL         string            `json:"url"`
+	Body        string            `json:"body"`
+	Time        time.Time         `json:"time"`
+	Title       string            `json:"title"`
+	Meta        map[string]string `json:"meta"`
+	Description string            `json:"description"`
+	Headers     map[string]string `json:"headers"`
+}
+
+// Exchange returns the exchange where event should be push
+func (msg *NewIndexEvent) Exchange() string {
+	return NewIndexExchange
+}