activitylogmq

README

pkg-activitylogmq

Shared Go library that carries IlonaPay activity-log (audit) events from any microservice to activity-log-service.

Services enqueue a JSON payload with one call; Watermill publishes it to a message broker; a consumer process forwards it over HTTP to POST /activity-logs.

EnqueueActivityLogCreate → activity-log.create → handleActivityLogCreate → POST /activity-logs

This module has no HTTP server and no database. Persistence and query APIs live in activity-log-service.

Overview

pkg-activitylogmq provides a small, dependency-light producer and consumer for application audit events. Producers call a single convenience function to enqueue an audit event; the library handles broker selection and publishing using Watermill. A consumer (also provided) reads messages and forwards them to the central activity-log-service over HTTP.

The library is designed to be embedded by any microservice that needs to emit activity logs without coupling that service to the log storage or HTTP latency.

Goals

• Provide a single-call, non-blocking API for services to publish audit events.
• Be broker-agnostic: support RabbitMQ / Amazon MQ (AMQPS), Google Cloud Pub/Sub, and Kafka.
• Fail open for missing broker configuration so services keep running when audit delivery is unavailable.
• Keep the library small and focused — no servers, no DBs, no schema ownership.
• Ensure producers and consumers share a common payload type so the log service and producers remain aligned.

Tech stack

• Language: Go (1.26+)
• Messaging: Watermill (publisher/subscriber abstraction)
• Brokers supported: RabbitMQ, Amazon MQ (RabbitMQ engine / AMQPS), Google Cloud Pub/Sub, Kafka
• Testing: go test, httptest for HTTP fakes, integration tests with RabbitMQ and Kafka (CI)
• Build / local infra: Docker Compose (for local RabbitMQ/Kafka during integration)

Features

• One-call publish via messaging.EnqueueActivityLogCreate
• Broker-agnostic: RabbitMQ, Amazon MQ (RabbitMQ/AMQPS), Google Cloud Pub/Sub, and Kafka
• Safe degradation when broker env is missing (host service keeps running)
• Non-blocking audits — publish does not wait on activity-log-service
• Shared clients.CreateBody payload across services

Requirements

• Go 1.26+

Installation

go get github.com/mawarpay/pkg-activitylogmq@latest

import (
    "github.com/mawarpay/pkg-activitylogmq/clients"
    "github.com/mawarpay/pkg-activitylogmq/messaging"
)

Quick start

package main

import (
    "context"
    "log"

    "github.com/mawarpay/pkg-activitylogmq/clients"
    "github.com/mawarpay/pkg-activitylogmq/messaging"
)

func main() {
    ctx := context.Background()
    if err := messaging.Init(ctx); err != nil {
        log.Fatal(err) // only broker wiring failures; missing config returns nil
    }
    defer messaging.Close()

    err := messaging.EnqueueActivityLogCreate(ctx, clients.CreateBody{
        LogName:     "auth",
        Description: "User logged in",
        SubjectType: "User",
        Event:       "login",
        SubjectID:   42,
        CauserType:  "User",
        CauserID:    42,
    })
    if err != nil {
        // typically log and continue — do not fail the business request
        log.Printf("activity log enqueue: %v", err)
    }
}

Low-level broker factory (without the messaging package):

import activitylogmq "github.com/mawarpay/pkg-activitylogmq"

cfg := activitylogmq.LoadConfig()
if cfg.Enabled() {
    pub, err := activitylogmq.NewPublisher(cfg, nil)
    // ...
    sub, err := activitylogmq.NewSubscriber(cfg, nil)
    // ...
}

Package layout

.
├── activitylogmq          # config + NewPublisher / NewSubscriber
├── clients/               # CreateBody + HTTP client for activity-log-service
├── messaging/             # Init, EnqueueActivityLogCreate, consumer handler
├── LICENSE
├── README.md
├── PRD.md                 # product requirements and known risks
└── PLAN.md                # gRPC + mTLS migration roadmap

Package	Import path	Role
activitylogmq	github.com/mawarpay/pkg-activitylogmq	Config and Watermill broker factory
clients	.../clients	CreateBody and HTTP Create
messaging	.../messaging	Process-wide Init / enqueue / consumer

API reference: pkg.go.dev/github.com/mawarpay/pkg-activitylogmq.

Configuration

All settings are environment variables. No broker configured → queue disabled (warn only).

Broker selection

Variable	Description
MESSAGE_BROKER	rabbitmq (amqp/rabbit), amazonmq (amazon_mq/amq), pubsub (google/gcp), or kafka. Auto-detected if unset.

Auto-detect order when MESSAGE_BROKER is unset: RabbitMQ URI present → PUBSUB_PROJECT_ID → KAFKA_BROKERS → Amazon MQ URI/host.

Topic / queue

Variable	Description
ACTIVITY_LOG_QUEUE	Topic / queue name (preferred)
ACTIVITY_LOG_TOPIC	Alias for queue name
PUBSUB_TOPIC	Pub/Sub fallback
KAFKA_TOPIC	Kafka fallback

Default when none are set: activity-log.create.

RabbitMQ

Variable	Description
RABBITMQ_URL	Full AMQP URI (overrides host-based config)
RABBITMQ_HOST	Host (required for host-based URI)
RABBITMQ_PORT	Default 5672
RABBITMQ_USER	Default guest
RABBITMQ_PASSWORD	Default guest
RABBITMQ_VHOST	Default /

Amazon MQ (RabbitMQ engine)

Uses the same Watermill AMQP adapter over AMQPS (TLS). ActiveMQ is not supported.

Variable	Description
AMAZONMQ_URL	Full amqps:// URI (overrides host-based config). Alias: AMAZON_MQ_URL.
AMAZONMQ_HOST	Broker endpoint hostname. Alias: AMAZON_MQ_HOST.
AMAZONMQ_PORT	Default 5671 (TLS) or 5672 when TLS is off. Alias: AMAZON_MQ_PORT.
AMAZONMQ_USER	Broker username. Alias: AMAZON_MQ_USER.
AMAZONMQ_PASSWORD	Broker password. Alias: AMAZON_MQ_PASSWORD.
AMAZONMQ_VHOST	Default /. Alias: AMAZON_MQ_VHOST.
AMAZONMQ_TLS	Default true (amqps). Set false for plain amqp (local/dev only). Alias: AMAZON_MQ_TLS.

environment:
  MESSAGE_BROKER: amazonmq
  AMAZONMQ_HOST: b-xxxxx.mq.us-east-1.amazonaws.com
  AMAZONMQ_USER: mquser
  AMAZONMQ_PASSWORD: secret
  ACTIVITY_LOG_QUEUE: activity-log.create

Google Cloud Pub/Sub

Variable	Description
PUBSUB_PROJECT_ID	GCP project ID

Auth uses standard GOOGLE_APPLICATION_CREDENTIALS.

Kafka

Variable	Description
KAFKA_BROKERS	Comma-separated broker addresses
KAFKA_CONSUMER_GROUP	Default activity-log-consumer

HTTP forward (consumer)

Variable	Description
ACTIVITY_LOG_SERVICE_URL	Base URL for POST /activity-logs (e.g. http://activity-log-service:8080)

Read once at package init into clients.ActivityLog. Consumers without this URL acknowledge and drop messages they receive.

messaging.Init starts a publisher and a consumer. Every service that calls Init binds the same durable queue activity-log.create and they compete for deliveries. See PRD.md and PLAN.md for migration and design notes.

Typical Compose snippet

environment:
  MESSAGE_BROKER: rabbitmq
  RABBITMQ_HOST: rabbitmq
  RABBITMQ_USER: guest
  RABBITMQ_PASSWORD: guest
  ACTIVITY_LOG_QUEUE: activity-log.create
  ACTIVITY_LOG_SERVICE_URL: http://activity-log-service:8080

Design philosophy

• Prefer async enqueue over synchronous HTTP on the request path.
• Fail open for configuration gaps; fail closed only on explicit wiring errors from Init.
• Keep this module a library — no routes, servers, or schema ownership.
• Share one payload type (CreateBody) so producers and the log service stay aligned.

Development

make help          # list targets
make check         # fmt-check + vet + build + test (race/cover) — matches CI unit steps
make test          # go test ./... -race -cover
make test-integration  # RabbitMQ + Kafka publish/subscribe (needs INTEGRATION=1)

make docker-up     # start RabbitMQ (:5672, UI :15672 guest/guest)
make docker-test   # build image and run go test ./... against Compose RabbitMQ
make docker-down   # stop Compose services

Unit tests use fakes and httptest; a live broker is not required for make test. CI starts RabbitMQ and Kafka service containers and runs make test-integration.

Docs

• PRD.md — requirements, config surface, known risks
• PLAN.md — mTLS + gRPC migration plan

Contributing

See CONTRIBUTING.md for setup, scope, API stability, and PR expectations.

License

MIT © 2026 mawarpay

Documentation

Overview

Package activitylogmq provides a Watermill broker factory for publishing and consuming IlonaPay activity-log (audit) messages.

Most application code should use the higher-level github.com/mawarpay/pkg-activitylogmq/messaging package, which wires publisher, subscriber, and the HTTP forwarder to activity-log-service in one Init call. This root package is for services that need direct control over Watermill publishers and subscribers.

Supported brokers

• RabbitMQ (AMQP)
• Amazon MQ for RabbitMQ (AMQPS)
• Google Cloud Pub/Sub
• Apache Kafka

Broker selection is driven by environment variables. Call LoadConfig then check Config.Enabled before constructing clients:

cfg := activitylogmq.LoadConfig()
if !cfg.Enabled() {
    // queue disabled; service continues without audit publishing
    return
}
pub, err := activitylogmq.NewPublisher(cfg, logger)
sub, err := activitylogmq.NewSubscriber(cfg, logger)

Configuration

See the module README for the full environment-variable reference. Key vars:

• MESSAGE_BROKER — rabbitmq | amazonmq | pubsub | kafka (auto-detected when unset)
• ACTIVITY_LOG_QUEUE — topic / queue name (default activity-log.create)
• RABBITMQ_URL or RABBITMQ_HOST — RabbitMQ connection
• AMAZONMQ_URL or AMAZONMQ_HOST — Amazon MQ for RabbitMQ (AMQPS, port 5671)
• PUBSUB_PROJECT_ID — Google Cloud project
• KAFKA_BROKERS — comma-separated Kafka addresses

Caveats

RabbitMQ and Amazon MQ use Watermill's durable queue config, which names the queue after the topic. Multiple subscribers on the same topic compete for deliveries. NewPublisher and NewSubscriber dial the broker; callers must Close the returned clients when finished.

Index

• func NewPublisher(cfg Config, logger watermill.LoggerAdapter) (message.Publisher, error)
• func NewSubscriber(cfg Config, logger watermill.LoggerAdapter) (message.Subscriber, error)
• type Broker
• type Config
  • func LoadConfig() Config
  • func (c Config) Enabled() bool

Examples

• Config.Enabled
• LoadConfig
• NewPublisher

Functions

func NewPublisher

func NewPublisher(cfg Config, logger watermill.LoggerAdapter) (message.Publisher, error)

NewPublisher creates a Watermill message.Publisher for the broker described by cfg.

cfg must be Config.Enabled; otherwise an error is returned and no connection is attempted. A nil logger is replaced with Watermill's standard logger.

The returned publisher dials the broker and must be closed by the caller when it is no longer needed. Errors wrapping dial or configuration failures are returned as-is from the underlying Watermill adapter.

Example

package main

import (
	"fmt"

	activitylogmq "github.com/mawarpay/pkg-activitylogmq"
)

func main() {
	_, err := activitylogmq.NewPublisher(activitylogmq.Config{
		Broker: activitylogmq.BrokerRabbitMQ,
	}, nil)
	fmt.Println(err != nil)
}

Output:
true

func NewSubscriber

func NewSubscriber(cfg Config, logger watermill.LoggerAdapter) (message.Subscriber, error)

NewSubscriber creates a Watermill message.Subscriber for the broker described by cfg.

cfg must be Config.Enabled; otherwise an error is returned and no connection is attempted. A nil logger is replaced with Watermill's standard logger.

For RabbitMQ and Amazon MQ, the queue name equals the topic (durable queue topology). For Kafka, cfg.KafkaConsumerGroup is used. The returned subscriber must be closed by the caller when finished.

Types

type Broker

type Broker string

Broker identifies the message transport selected by LoadConfig.

const (
	// BrokerRabbitMQ selects RabbitMQ / AMQP. Aliases for MESSAGE_BROKER:
	// "rabbitmq", "amqp", "rabbit".
	BrokerRabbitMQ Broker = "rabbitmq"

	// BrokerAmazonMQ selects Amazon MQ for RabbitMQ over AMQPS (TLS).
	// It reuses the Watermill AMQP adapter with an amqps:// URI.
	// Aliases for MESSAGE_BROKER: "amazonmq", "amazon_mq", "amazon-mq", "amq".
	BrokerAmazonMQ Broker = "amazonmq"

	// BrokerPubSub selects Google Cloud Pub/Sub. Aliases for MESSAGE_BROKER:
	// "pubsub", "google", "gcp", "google_pubsub".
	BrokerPubSub Broker = "pubsub"

	// BrokerKafka selects Apache Kafka.
	BrokerKafka Broker = "kafka"
)

type Config

type Config struct {
	// Broker is the selected transport. Empty when no broker could be detected.
	Broker Broker

	// Topic is the queue or topic name used for publish and subscribe.
	// Defaults to "activity-log.create" when no topic env var is set.
	Topic string

	// RabbitMQURI is the AMQP/AMQPS connection URI when Broker is
	// BrokerRabbitMQ or BrokerAmazonMQ.
	RabbitMQURI string

	// PubSubProjectID is the GCP project ID when Broker is BrokerPubSub.
	PubSubProjectID string

	// KafkaBrokers is the list of Kafka bootstrap addresses when Broker is
	// BrokerKafka.
	KafkaBrokers []string

	// KafkaConsumerGroup is the consumer group for Kafka subscribers.
	// Defaults to "activity-log-consumer".
	KafkaConsumerGroup string
}

Config holds broker connection settings loaded from the environment by LoadConfig. Pass an Enabled config to NewPublisher or NewSubscriber.

func LoadConfig

func LoadConfig() Config

LoadConfig reads MESSAGE_BROKER (or auto-detects the broker) and broker-specific environment variables into a Config.

Topic resolution order: ACTIVITY_LOG_QUEUE, ACTIVITY_LOG_TOPIC, PUBSUB_TOPIC, KAFKA_TOPIC, then "activity-log.create".

Broker auto-detect order when MESSAGE_BROKER is unset: RabbitMQ URI present, then PUBSUB_PROJECT_ID, then KAFKA_BROKERS, then Amazon MQ URI/host.

LoadConfig never returns an error; call Config.Enabled to determine whether publishing and subscribing can proceed.

Example

package main

import (
	"fmt"

	activitylogmq "github.com/mawarpay/pkg-activitylogmq"
)

func main() {
	// LoadConfig reads MESSAGE_BROKER and broker-specific env vars.
	// Check Enabled before constructing publishers or subscribers.
	cfg := activitylogmq.LoadConfig()
	if !cfg.Enabled() {
		fmt.Println("broker disabled")
		return
	}

	pub, err := activitylogmq.NewPublisher(cfg, nil)
	if err != nil {
		fmt.Println("publisher:", err)
		return
	}
	defer pub.Close()

	sub, err := activitylogmq.NewSubscriber(cfg, nil)
	if err != nil {
		fmt.Println("subscriber:", err)
		return
	}
	defer sub.Close()
}

func (Config) Enabled

func (c Config) Enabled() bool

Enabled reports whether Broker is set and the corresponding connection settings are present (AMQP URI, Pub/Sub project ID, or Kafka brokers). A disabled config must not be passed to NewPublisher or NewSubscriber.

Example

package main

import (
	"fmt"

	activitylogmq "github.com/mawarpay/pkg-activitylogmq"
)

func main() {
	enabled := activitylogmq.Config{
		Broker:      activitylogmq.BrokerRabbitMQ,
		RabbitMQURI: "amqp://guest:guest@localhost:5672/",
	}
	disabled := activitylogmq.Config{
		Broker: activitylogmq.BrokerRabbitMQ,
	}

	fmt.Println(enabled.Enabled())
	fmt.Println(disabled.Enabled())
}

Output:
true
false