Add search indexing

_embed/templates/admin-search/index.html

@@ -0,0 +1,11 @@
+{{- $parent := .QueryParam "parent" -}}
+{{- $name := .QueryParam "name" -}}
+{{- $stateID := .QueryParam "stateId" -}}
+
+<div class="page" hx-get="/admin/tags/index" hx-trigger="refreshPage from:window">
+
+	{{template "menubar" .}}
+
+	<button hx-post="/admin/index-all-streams" hx-swap="none" hx-push-url="false">Index All Streams</button>
+
+</div>

_embed/templates/admin-search/template.hjson

@@ -0,0 +1,13 @@
+{
+	templateId:admin-search
+	templateRole:admin
+	model:search
+	extends: ["admin-common"]
+	containedBy:["admin"]
+	label:Search
+	description: Manage Search Engine Settings
+
+	actions: {
+		index: {do:"view-html"}
+	}
+}

consumer/consumer.go

@@ -28,6 +28,9 @@ func (consumer Consumer) Run(name string, args map[string]any) queue.Result {
 	case "CreateWebSubFollower":
 		return WithFactory(consumer.serverFactory, args, CreateWebSubFollower)
 
+	case "IndexAllStreams":
+		return WithFactory(consumer.serverFactory, args, IndexAllStreams)
+
 	case "MakeStreamArchive":
 		return WithStream(consumer.serverFactory, args, MakeStreamArchive)
 

consumer/indexAllStreams.go

@@ -0,0 +1,37 @@
+package consumer
+
+import (
+	"github.com/EmissarySocial/emissary/domain"
+	"github.com/benpate/derp"
+	"github.com/benpate/remote"
+	"github.com/benpate/rosetta/mapof"
+	"github.com/benpate/turbine/queue"
+	"github.com/rs/zerolog/log"
+)
+
+func IndexAllStreams(factory *domain.Factory, args mapof.Any) queue.Result {
+
+	const location = "consumer.IndexAllStreams"
+
+	streamService := factory.Stream()
+
+	allStreams, err := streamService.RangeAll()
+
+	if err != nil {
+		return queue.Error(derp.Wrap(err, location, "Error retrieving Streams"))
+	}
+
+	for summary := range allStreams {
+
+		log.Debug().Str("url", summary.URL).Msg("Indexing Stream")
+		transaction := remote.Post(summary.URL + "/search-index")
+
+		if err := transaction.Send(); err != nil {
+			if !derp.IsClientError(err) {
+				return queue.Error(derp.Wrap(err, location, "Error sending request"))
+			}
+		}
+	}
+
+	return queue.Success()
+}

handler/admin.go

@@ -92,6 +92,9 @@ func buildAdmin_GetBuilder(factory *domain.Factory, ctx *steranko.Context, templ
 	case "domain":
 		return build.NewDomain(factory, ctx.Request(), ctx.Response(), template, actionID)
 
+	case "search":
+		return build.NewDomain(factory, ctx.Request(), ctx.Response(), template, actionID)
+
 	case "syndication":
 		return build.NewSyndication(factory, ctx.Request(), ctx.Response(), template, actionID)
 

handler/search.go

@@ -0,0 +1,36 @@
+package handler
+
+import (
+	"net/http"
+
+	"github.com/EmissarySocial/emissary/domain"
+	"github.com/benpate/derp"
+	"github.com/benpate/rosetta/mapof"
+	"github.com/benpate/steranko"
+	"github.com/benpate/turbine/queue"
+)
+
+// IndexAllStreams is a handler function that triggers the IndexAllStreams queue task.
+// It can only be called by an authenticated administrator.
+func IndexAllStreams(ctx *steranko.Context, factory *domain.Factory) error {
+
+	// Verify that this is an Administrator
+	authorization := getAuthorization(ctx)
+
+	if !authorization.DomainOwner {
+		return derp.NewForbiddenError("handler.IndexAllStreams", "Only administrators can call this method")
+	}
+
+	// Create the Index task
+	task := queue.NewTask("IndexAllStreams", mapof.Any{
+		"host": ctx.Request().Host,
+	})
+
+	// Execute the task in the background
+	if err := factory.Queue().Publish(task); err != nil {
+		return derp.Wrap(err, "handler.IndexAllStreams", "Error publishing task")
+	}
+
+	// Success.
+	return ctx.NoContent(http.StatusOK)
+}

server.go

@@ -344,6 +344,7 @@ func makeStandardRoutes(factory *server.Factory, e *echo.Echo) {
 	e.POST("/admin/:param1/:param2", handler.PostAdmin(factory), mw.Owner)
 	e.GET("/admin/:param1/:param2/:param3", handler.GetAdmin(factory), mw.Owner)
 	e.POST("/admin/:param1/:param2/:param3", handler.PostAdmin(factory), mw.Owner)
+	e.POST("/admin/index-all-streams", handler.WithFactory(factory, handler.IndexAllStreams), mw.Owner)
 
 	// OAuth Client Connections
 	e.GET("/oauth/clients/:provider", handler.GetOAuth(factory), mw.Owner)

service/follower.go

@@ -77,6 +77,7 @@ func (service *Follower) List(criteria exp.Expression, options ...option.Option)
 	return service.collection.Iterator(notDeleted(criteria), options...)
 }
 
+// Range returns a Go 1.23 RangeFunc that iterates over the Followers who match the provided criteria
 func (service *Follower) Range(criteria exp.Expression, options ...option.Option) (iter.Seq[model.Follower], error) {
 
 	iter, err := service.List(criteria, options...)

service/stream.go

@@ -2,6 +2,7 @@ package service
 
 import (
 	"context"
+	"iter"
 	"net/url"
 	"strings"
 	"time"
@@ -177,6 +178,30 @@ func (service *Stream) QuerySummary(criteria exp.Expression, options ...option.O
 	return result, err
 }
 
+// Range returns a Go 1.23 RangeFunc that iterates over the Streams that match the provided criteria
+func (service *Stream) Range(criteria exp.Expression, options ...option.Option) (iter.Seq[model.Stream], error) {
+
+	iter, err := service.List(criteria, options...)
+
+	if err != nil {
+		return nil, derp.Wrap(err, "service.Stream.Range", "Error creating iterator", criteria)
+	}
+
+	return RangeFunc(iter, model.NewStream), nil
+}
+
+// RangeSummary returns a Go 1.23 RangeFunc that iterates over the Stream Summaries that match the provided criteria
+func (service *Stream) RangeSummary(criteria exp.Expression, options ...option.Option) (iter.Seq[model.StreamSummary], error) {
+
+	iter, err := service.List(criteria, options...)
+
+	if err != nil {
+		return nil, derp.Wrap(err, "service.Stream.Range", "Error creating iterator", criteria)
+	}
+
+	return RangeFunc(iter, model.NewStreamSummary), nil
+}
+
 // List returns an iterator containing all of the Streams that match the provided criteria
 func (service *Stream) List(criteria exp.Expression, options ...option.Option) (data.Iterator, error) {
 	return service.collection.Iterator(notDeleted(criteria), options...)
@@ -422,6 +447,10 @@ func (service *Stream) Schema() schema.Schema {
  * Custom Queries
  ******************************************/
 
+func (service *Stream) RangeAll() (iter.Seq[model.StreamSummary], error) {
+	return service.RangeSummary(exp.All())
+}
+
 // ListNavigation returns all Streams of type FOLDER at the top of the hierarchy
 func (service *Stream) ListNavigation() (data.Iterator, error) {
 	return service.List(