Second checkpoint

internal/messaging/rabbitmq/connection.go

@@ -0,0 +1,80 @@
+// Package rabbitmq provides the RabbitMQ connection, a publisher, and consumers
+// used by the worker. A single durable topic exchange is declared on connect;
+// each consumer declares and binds its own queue.
+package rabbitmq
+
+import (
+	"fmt"
+
+	"gis/internal/config"
+
+	amqp "github.com/rabbitmq/amqp091-go"
+)
+
+// Connection wraps an AMQP connection and a dedicated publishing channel, and
+// owns the topic exchange.
+type Connection struct {
+	conn     *amqp.Connection
+	pubCh    *amqp.Channel
+	exchange string
+}
+
+// Connect dials RabbitMQ, opens a publishing channel, and declares the exchange.
+func Connect(cfg config.RabbitMQConfig) (*Connection, error) {
+	conn, err := amqp.Dial(cfg.URL)
+	if err != nil {
+		return nil, fmt.Errorf("dial rabbitmq: %w", err)
+	}
+
+	ch, err := conn.Channel()
+	if err != nil {
+		conn.Close()
+		return nil, fmt.Errorf("open channel: %w", err)
+	}
+
+	if err := ch.ExchangeDeclare(
+		cfg.Exchange, amqp.ExchangeTopic,
+		true,  // durable
+		false, // auto-deleted
+		false, // internal
+		false, // no-wait
+		nil,
+	); err != nil {
+		ch.Close()
+		conn.Close()
+		return nil, fmt.Errorf("declare exchange: %w", err)
+	}
+
+	return &Connection{conn: conn, pubCh: ch, exchange: cfg.Exchange}, nil
+}
+
+// Exchange returns the topic exchange name.
+func (c *Connection) Exchange() string { return c.exchange }
+
+// publishChannel returns the shared publishing channel.
+func (c *Connection) publishChannel() *amqp.Channel { return c.pubCh }
+
+// openChannel opens a fresh channel (each consumer uses its own).
+func (c *Connection) openChannel() (*amqp.Channel, error) { return c.conn.Channel() }
+
+// Ping reports whether the connection is still open (used by readiness checks).
+func (c *Connection) Ping() error {
+	if c.conn.IsClosed() {
+		return fmt.Errorf("rabbitmq connection closed")
+	}
+	return nil
+}
+
+// Close tears down the publishing channel and the connection.
+func (c *Connection) Close() error {
+	var chErr error
+	if c.pubCh != nil {
+		chErr = c.pubCh.Close()
+	}
+	if c.conn != nil {
+		if err := c.conn.Close(); err != nil {
+			return err
+		}
+	}
+	return chErr
+}

internal/messaging/rabbitmq/consumer.go

@@ -0,0 +1,113 @@
+package rabbitmq
+
+import (
+	"context"
+	"fmt"
+	"log/slog"
+
+	amqp "github.com/rabbitmq/amqp091-go"
+)
+
+// Handler processes a single delivery. Returning nil acks the message; returning
+// an error nacks it (without requeue, to avoid poison-message loops).
+type Handler interface {
+	Handle(ctx context.Context, d amqp.Delivery) error
+}
+
+// Consumer declares a durable queue bound to the exchange by routing key and
+// dispatches deliveries to a Handler. Each Consumer uses its own channel.
+type Consumer struct {
+	conn       *Connection
+	queue      string
+	bindingKey string
+	tag        string
+	handler    Handler
+	log        *slog.Logger
+}
+
+// NewConsumer returns a Consumer for the given queue and routing-key binding.
+func NewConsumer(conn *Connection, queue, bindingKey, tag string, handler Handler, log *slog.Logger) *Consumer {
+	return &Consumer{
+		conn:       conn,
+		queue:      queue,
+		bindingKey: bindingKey,
+		tag:        tag,
+		handler:    handler,
+		log:        log,
+	}
+}
+
+// Run declares/binds the queue and consumes until ctx is cancelled or the
+// delivery channel closes. It uses manual acknowledgement.
+func (c *Consumer) Run(ctx context.Context) error {
+	ch, err := c.conn.openChannel()
+	if err != nil {
+		return fmt.Errorf("open channel: %w", err)
+	}
+	defer ch.Close()
+
+	if err := c.setup(ch); err != nil {
+		return err
+	}
+
+	deliveries, err := ch.Consume(
+		c.queue, c.tag,
+		false, // auto-ack: we ack manually
+		false, // exclusive
+		false, // no-local
+		false, // no-wait
+		nil,
+	)
+	if err != nil {
+		return fmt.Errorf("start consume: %w", err)
+	}
+
+	c.log.Info("consumer started", "queue", c.queue, "binding", c.bindingKey, "tag", c.tag)
+
+	for {
+		select {
+		case <-ctx.Done():
+			if err := ch.Cancel(c.tag, false); err != nil {
+				c.log.Warn("cancel consumer", "error", err)
+			}
+			c.log.Info("consumer stopped", "tag", c.tag)
+			return ctx.Err()
+
+		case d, ok := <-deliveries:
+			if !ok {
+				return fmt.Errorf("delivery channel closed for queue %q", c.queue)
+			}
+			c.dispatch(ctx, d)
+		}
+	}
+}
+
+func (c *Consumer) setup(ch *amqp.Channel) error {
+	if _, err := ch.QueueDeclare(
+		c.queue,
+		true,  // durable
+		false, // auto-delete
+		false, // exclusive
+		false, // no-wait
+		nil,
+	); err != nil {
+		return fmt.Errorf("declare queue %q: %w", c.queue, err)
+	}
+	if err := ch.QueueBind(c.queue, c.bindingKey, c.conn.Exchange(), false, nil); err != nil {
+		return fmt.Errorf("bind queue %q: %w", c.queue, err)
+	}
+	return nil
+}
+
+func (c *Consumer) dispatch(ctx context.Context, d amqp.Delivery) {
+	if err := c.handler.Handle(ctx, d); err != nil {
+		c.log.Error("handle delivery", "routing_key", d.RoutingKey, "error", err)
+		if nackErr := d.Nack(false, false); nackErr != nil {
+			c.log.Error("nack delivery", "error", nackErr)
+		}
+		return
+	}
+	if ackErr := d.Ack(false); ackErr != nil {
+		c.log.Error("ack delivery", "error", ackErr)
+	}
+}

internal/messaging/rabbitmq/example_consumer.go

@@ -0,0 +1,74 @@
+package rabbitmq
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"log/slog"
+	"time"
+
+	amqp "github.com/rabbitmq/amqp091-go"
+)
+
+const (
+	// ExampleRoutingKey is the routing key used by the generic example flow.
+	ExampleRoutingKey = "example.created"
+	// ExampleBindingKey binds the example queue to example.* routing keys.
+	ExampleBindingKey = "example.#"
+)
+
+// EventRecorder persists a received event. It is the seam between the messaging
+// layer and storage for the example flow; a real domain would call its own
+// service instead.
+type EventRecorder interface {
+	Record(ctx context.Context, kind string, payload json.RawMessage) error
+}
+
+// ExampleMessage is the payload published and consumed by the scaffold.
+type ExampleMessage struct {
+	Kind      string    `json:"kind"`
+	Message   string    `json:"message"`
+	EmittedAt time.Time `json:"emitted_at"`
+}
+
+// ExampleHandler is a generic consumer that records every delivery into the
+// events table. It demonstrates the messaging -> storage path; delete it when a
+// real async use case replaces the scaffold.
+type ExampleHandler struct {
+	recorder EventRecorder
+	log      *slog.Logger
+}
+
+// NewExampleHandler returns an ExampleHandler.
+func NewExampleHandler(recorder EventRecorder, log *slog.Logger) *ExampleHandler {
+	return &ExampleHandler{recorder: recorder, log: log}
+}
+
+// Handle decodes the delivery (best-effort) and records it.
+func (h *ExampleHandler) Handle(ctx context.Context, d amqp.Delivery) error {
+	var msg ExampleMessage
+	if err := json.Unmarshal(d.Body, &msg); err != nil {
+		return fmt.Errorf("decode example message: %w", err)
+	}
+
+	h.log.Info("received example message",
+		"routing_key", d.RoutingKey, "kind", msg.Kind, "message", msg.Message)
+
+	if err := h.recorder.Record(ctx, "example", d.Body); err != nil {
+		return fmt.Errorf("record event: %w", err)
+	}
+	return nil
+}
+
+// PublishExample emits a single example message; used by `gis worker --publish-example`.
+func PublishExample(ctx context.Context, pub *Publisher) error {
+	body, err := json.Marshal(ExampleMessage{
+		Kind:      "example",
+		Message:   "hello from gis worker",
+		EmittedAt: time.Now().UTC(),
+	})
+	if err != nil {
+		return err
+	}
+	return pub.Publish(ctx, ExampleRoutingKey, body)
+}

internal/messaging/rabbitmq/parse_consumer.go

@@ -0,0 +1,118 @@
+package rabbitmq
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"log/slog"
+
+	"github.com/google/uuid"
+	amqp "github.com/rabbitmq/amqp091-go"
+)
+
+const (
+	// DatasetParseRoutingKey routes dataset attribute-table parse jobs.
+	DatasetParseRoutingKey = "dataset.parse"
+	// DatasetParseQueue receives dataset parse jobs.
+	DatasetParseQueue = "gis.datasets.parse"
+	// DatasetPropertiesRoutingKey routes plain-vector attribute-table extraction.
+	DatasetPropertiesRoutingKey = "dataset.properties"
+	// DatasetPropertiesQueue receives plain-vector properties jobs.
+	DatasetPropertiesQueue = "gis.datasets.properties"
+	// DatasetExtractRoutingKey routes dataset extraction (unpivot) jobs.
+	DatasetExtractRoutingKey = "dataset.extract"
+	// DatasetExtractQueue receives dataset extraction jobs.
+	DatasetExtractQueue = "gis.datasets.extract"
+	// DatasetConvertRoutingKey routes raster COG-conversion jobs.
+	DatasetConvertRoutingKey = "dataset.cog"
+	// DatasetConvertQueue receives raster COG-conversion jobs.
+	DatasetConvertQueue = "gis.datasets.cog"
+)
+
+// DatasetJob is the message body for a dataset job (parse or extract).
+type DatasetJob struct {
+	DatasetID uuid.UUID `json:"dataset_id"`
+}
+
+// DatasetJobPublisher publishes dataset parse/extract jobs; it implements
+// service.JobEnqueuer.
+type DatasetJobPublisher struct {
+	pub *Publisher
+}
+
+// NewDatasetJobPublisher returns a DatasetJobPublisher.
+func NewDatasetJobPublisher(pub *Publisher) *DatasetJobPublisher {
+	return &DatasetJobPublisher{pub: pub}
+}
+
+// EnqueueParse publishes a parse job for the given dataset.
+func (p *DatasetJobPublisher) EnqueueParse(ctx context.Context, datasetID uuid.UUID) error {
+	return p.publish(ctx, DatasetParseRoutingKey, datasetID)
+}
+
+// EnqueueProperties publishes a plain-vector properties-extraction job.
+func (p *DatasetJobPublisher) EnqueueProperties(ctx context.Context, datasetID uuid.UUID) error {
+	return p.publish(ctx, DatasetPropertiesRoutingKey, datasetID)
+}
+
+// EnqueueExtract publishes an extraction job for the given dataset.
+func (p *DatasetJobPublisher) EnqueueExtract(ctx context.Context, datasetID uuid.UUID) error {
+	return p.publish(ctx, DatasetExtractRoutingKey, datasetID)
+}
+
+// EnqueueConvert publishes a raster COG-conversion job for the given dataset.
+func (p *DatasetJobPublisher) EnqueueConvert(ctx context.Context, datasetID uuid.UUID) error {
+	return p.publish(ctx, DatasetConvertRoutingKey, datasetID)
+}
+
+func (p *DatasetJobPublisher) publish(ctx context.Context, routingKey string, datasetID uuid.UUID) error {
+	body, err := json.Marshal(DatasetJob{DatasetID: datasetID})
+	if err != nil {
+		return err
+	}
+	return p.pub.Publish(ctx, routingKey, body)
+}
+
+// DatasetProcessor runs the async dataset jobs; implemented by the service.
+type DatasetProcessor interface {
+	Parse(ctx context.Context, datasetID uuid.UUID) error
+	ExtractProperties(ctx context.Context, datasetID uuid.UUID) error
+	Extract(ctx context.Context, datasetID uuid.UUID) error
+	ConvertToCOG(ctx context.Context, datasetID uuid.UUID) error
+}
+
+// jobHandler dispatches a dataset job to one processor function.
+type jobHandler struct {
+	name string
+	fn   func(ctx context.Context, id uuid.UUID) error
+	log  *slog.Logger
+}
+
+func (h jobHandler) Handle(ctx context.Context, d amqp.Delivery) error {
+	var job DatasetJob
+	if err := json.Unmarshal(d.Body, &job); err != nil {
+		return fmt.Errorf("decode %s job: %w", h.name, err)
+	}
+	h.log.Info("processing dataset "+h.name, "dataset_id", job.DatasetID)
+	return h.fn(ctx, job.DatasetID)
+}
+
+// NewParseHandler returns a handler that parses datasets.
+func NewParseHandler(p DatasetProcessor, log *slog.Logger) Handler {
+	return jobHandler{name: "parse", fn: p.Parse, log: log}
+}
+
+// NewPropertiesHandler returns a handler that extracts a plain vector's table.
+func NewPropertiesHandler(p DatasetProcessor, log *slog.Logger) Handler {
+	return jobHandler{name: "properties", fn: p.ExtractProperties, log: log}
+}
+
+// NewExtractHandler returns a handler that extracts (unpivots) datasets.
+func NewExtractHandler(p DatasetProcessor, log *slog.Logger) Handler {
+	return jobHandler{name: "extract", fn: p.Extract, log: log}
+}
+
+// NewConvertHandler returns a handler that converts rasters to COGs.
+func NewConvertHandler(p DatasetProcessor, log *slog.Logger) Handler {
+	return jobHandler{name: "convert", fn: p.ConvertToCOG, log: log}
+}

internal/messaging/rabbitmq/publisher.go

@@ -0,0 +1,37 @@
+package rabbitmq
+
+import (
+	"context"
+	"fmt"
+
+	amqp "github.com/rabbitmq/amqp091-go"
+)
+
+// Publisher publishes messages to the connection's exchange.
+type Publisher struct {
+	conn *Connection
+}
+
+// NewPublisher returns a Publisher bound to the given connection.
+func NewPublisher(conn *Connection) *Publisher {
+	return &Publisher{conn: conn}
+}
+
+// Publish sends a JSON-encoded body to the exchange using the given routing key.
+func (p *Publisher) Publish(ctx context.Context, routingKey string, body []byte) error {
+	err := p.conn.publishChannel().PublishWithContext(ctx,
+		p.conn.Exchange(),
+		routingKey,
+		false, // mandatory
+		false, // immediate
+		amqp.Publishing{
+			ContentType:  "application/json",
+			DeliveryMode: amqp.Persistent,
+			Body:         body,
+		},
+	)
+	if err != nil {
+		return fmt.Errorf("publish to %q: %w", routingKey, err)
+	}
+	return nil
+}