events

import "github.com/prescriptoo/prescriptoo-api/pkg/events"

Package events provides a lightweight event bus for inter-module communication. The in-process implementation can be swapped with RabbitMQ/NATS when extracting to microservices.

Index

• Constants
• type Bus
• type Event
• type InMemoryBus
  • func NewInMemoryBus() *InMemoryBus
  • func (b *InMemoryBus) Publish(_ context.Context, event Event) error
  • func (b *InMemoryBus) Subscribe(eventType string, handler func(Event)) error
• type RabbitMQBus
  • func NewRabbitMQBus(amqpURL string) (*RabbitMQBus, error)
  • func NewRabbitMQBusWithExchange(amqpURL, exchange string) (*RabbitMQBus, error)
  • func (b *RabbitMQBus) Close() error
  • func (b *RabbitMQBus) Publish(ctx context.Context, event Event) error
  • func (b *RabbitMQBus) Subscribe(eventType string, handler func(Event)) error

Constants

Identity events

const (
    // UserRegistered is published when a new user registers.
    // Payload: user_id, email, role
    UserRegistered = "user.registered"

    // UserLoggedIn is published when a user successfully logs in.
    // Payload: user_id, email
    UserLoggedIn = "user.logged_in"

    // DoctorProfileCreated is published when a doctor profile is created.
    // Payload: user_id, license_number
    DoctorProfileCreated = "doctor.profile.created"

    // StaffProfileCreated is published when a staff profile is created.
    // Payload: user_id, organization_id
    StaffProfileCreated = "staff.profile.created"
)

Organization events

const (
    // OrganizationCreated is published when a new organization is created.
    // Payload: organization_id, name, admin_user_id
    OrganizationCreated = "organization.created"

    // OrganizationSuspended is published when an organization is suspended.
    // Payload: organization_id
    OrganizationSuspended = "organization.suspended"

    // OrganizationActivated is published when an organization is activated.
    // Payload: organization_id
    OrganizationActivated = "organization.activated"

    // OrganizationUpdated is published after organization details change.
    OrganizationUpdated = "organization.updated"

    // OrganizationDeleted is published after an organization is soft-deleted.
    OrganizationDeleted = "organization.deleted"

    // OrganizationRestored is published after a deleted organization is restored.
    OrganizationRestored = "organization.restored"

    // OrganizationMemberCreated is published after an active member is created.
    OrganizationMemberCreated = "organization.member_created"

    // MemberInvited is published when a member is invited.
    // Payload: organization_id, user_id, role
    MemberInvited = "member.invited"

    // MemberJoined is published when a member accepts invitation.
    // Payload: organization_id, user_id
    MemberJoined = "member.joined"

    // MemberRemoved is published when a member is removed.
    // Payload: organization_id, user_id
    MemberRemoved = "member.removed"

    // TeamCreated is published when a team is created.
    // Payload: organization_id, team_id, name
    TeamCreated = "team.created"
)

Subscription events

const (
    // SubscriptionCreated is published when a subscription is created.
    // Payload: organization_id, plan_id, subscription_id
    SubscriptionCreated = "subscription.created"

    // SubscriptionCanceled is published when a subscription is canceled.
    // Payload: subscription_id, organization_id
    SubscriptionCanceled = "subscription.canceled"

    // SubscriptionRenewed is published when a subscription is renewed.
    // Payload: subscription_id, organization_id, period_end
    SubscriptionRenewed = "subscription.renewed"

    // BillingPaymentSucceeded is published when payment succeeds.
    // Payload: subscription_id, amount_cents
    BillingPaymentSucceeded = "billing.payment.succeeded"

    // BillingPaymentFailed is published when payment fails.
    // Payload: subscription_id, amount_cents
    BillingPaymentFailed = "billing.payment.failed"
)

Prescription events

const (
    // PrescriptionCreated is published when a prescription is created.
    // Payload: prescription_id, doctor_id, organization_id
    PrescriptionCreated = "prescription.created"

    // PrescriptionIssued is published when a prescription is issued.
    // Payload: prescription_id, doctor_id, patient_name
    PrescriptionIssued = "prescription.issued"

    // PrescriptionCanceled is published when a prescription is canceled.
    // Payload: prescription_id, doctor_id
    PrescriptionCanceled = "prescription.canceled"
)

Admin events

const (
    // AdminUserCreated is published after an identity is granted admin access.
    AdminUserCreated = "admin.user_created"

    // DoctorVerificationSubmitted is published when a doctor submits verification.
    // Payload: verification_id, doctor_profile_id, user_id
    DoctorVerificationSubmitted = "doctor.verification.submitted"

    // DoctorVerified is published when a doctor is verified.
    // Payload: verification_id, doctor_profile_id, admin_user_id
    DoctorVerified = "doctor.verified"

    // DoctorRejected is published when a doctor verification is rejected.
    // Payload: verification_id, doctor_profile_id, reason
    DoctorRejected = "doctor.rejected"
)

Notification events (for tracking)

const (
    // NotificationQueued is published when a notification is queued.
    // Payload: notification_id, channel, recipient
    NotificationQueued = "notification.queued"

    // NotificationSent is published when a notification is sent.
    // Payload: notification_id, channel, recipient
    NotificationSent = "notification.sent"

    // NotificationFailed is published when a notification fails.
    // Payload: notification_id, error
    NotificationFailed = "notification.failed"
)

const (
    // DefaultRabbitMQExchange is the topic exchange used by RabbitMQBus.
    DefaultRabbitMQExchange = "prescriptoo.events"
)

type Bus

Bus is the interface modules depend on for publishing and subscribing to events. Implementations: InMemoryBus and RabbitMQBus.

type Bus interface {
    // Publish sends an event to all registered subscribers.
    Publish(ctx context.Context, event Event) error

    // Subscribe registers a handler for events of the given type.
    Subscribe(eventType string, handler func(Event)) error
}

type Event

Event represents a domain event for cross-module communication.

type Event struct {
    Type       string            `json:"type"`
    Payload    map[string]string `json:"payload"`
    OccurredAt time.Time         `json:"occurred_at"`
}

type InMemoryBus

InMemoryBus is a simple in-process event bus implementation. This is the default for the modular monolith — swap with RabbitMQBus when extracting services.

type InMemoryBus struct {
    // contains filtered or unexported fields
}

func NewInMemoryBus

func NewInMemoryBus() *InMemoryBus

NewInMemoryBus creates a new in-memory event bus.

func (*InMemoryBus) Publish

func (b *InMemoryBus) Publish(_ context.Context, event Event) error

Publish sends an event to all registered subscribers synchronously.

Example

InMemoryBus invokes each matching subscriber before Publish returns.

package main

import (
	"context"
	"fmt"

	"github.com/prescriptoo/prescriptoo-api/pkg/events"
)

func main() {
	bus := events.NewInMemoryBus()
	if err := bus.Subscribe(events.UserRegistered, func(event events.Event) {
		fmt.Println("Registered:", event.Payload["user_id"])
	}); err != nil {
		panic(err)
	}
	if err := bus.Publish(context.Background(), events.Event{
		Type: events.UserRegistered, Payload: map[string]string{"user_id": "example-user"},
	}); err != nil {
		panic(err)
	}
}

Output

Registered: example-user

func (*InMemoryBus) Subscribe

func (b *InMemoryBus) Subscribe(eventType string, handler func(Event)) error

Subscribe registers a handler for events of the given type.

type RabbitMQBus

RabbitMQBus distributes events through a RabbitMQ topic exchange.

Every Subscribe call creates its own server-named queue. This preserves the InMemoryBus fan-out semantics: every registered handler receives every event matching its event type. Subscriptions are intentionally tied to the bus lifetime and are removed by RabbitMQ when the connection closes.

type RabbitMQBus struct {
    // contains filtered or unexported fields
}

func NewRabbitMQBus

func NewRabbitMQBus(amqpURL string) (*RabbitMQBus, error)

NewRabbitMQBus connects to RabbitMQ and creates a bus using DefaultRabbitMQExchange.

func NewRabbitMQBusWithExchange

func NewRabbitMQBusWithExchange(amqpURL, exchange string) (*RabbitMQBus, error)

NewRabbitMQBusWithExchange connects to RabbitMQ and creates a bus using the supplied exchange name. The exchange is durable and is not auto-deleted.

func (*RabbitMQBus) Close

func (b *RabbitMQBus) Close() error

Close stops all subscriptions and closes the RabbitMQ connection. It is safe to call Close more than once.

func (*RabbitMQBus) Publish

func (b *RabbitMQBus) Publish(ctx context.Context, event Event) error

Publish serializes event as JSON and routes it using its event type.

func (*RabbitMQBus) Subscribe

func (b *RabbitMQBus) Subscribe(eventType string, handler func(Event)) error

Subscribe registers a handler for an event type. Deliveries are processed in order for this subscription and acknowledged only after the handler returns.

Generated by gomarkdoc