クリーンアーキテクチャ設計書¶

// Domain Entities
package domain

import "time"

// Notification represents a notification to be sent
type Notification struct {
    NotificationID   string                // UUID
    UserID           string                // User to receive notification
    Title            string
    Body             string
    NotificationType NotificationType      // MESSAGE_RECEIVED, GROUP_CREATED, etc.
    DeliveryChannels []DeliveryChannel     // [PUSH, EMAIL, IN_APP]
    Priority         NotificationPriority  // HIGH, NORMAL, LOW
    Status           string                // PENDING, SENT, FAILED
    CreatedAt        time.Time
    ExpiresAt        time.Time
    Data             map[string]string     // Extra context
}

// DeviceToken represents a device push notification token
type DeviceToken struct {
    DeviceTokenID string       // UUID
    UserID        string
    DeviceID      string
    Platform      DevicePlatform
    TokenValue    string       // FCM token
    IsValid       bool
    LastUsedAt    time.Time
    RegisteredAt  time.Time
    ExpiresAt     time.Time
}

func (dt *DeviceToken) IsExpired(now time.Time) bool {
    return now.After(dt.ExpiresAt)
}

// NotificationPreference represents user's notification settings
type NotificationPreference struct {
    PreferenceID       string             // UUID
    UserID             string
    NotificationType   NotificationType
    EmailFrequency     string             // DAILY, WEEKLY, NEVER
    PushEnabled        bool
    InAppEnabled       bool
    QuietHoursStart    int                // 0-23
    QuietHoursEnd      int                // 0-23
    UpdatedAt          time.Time
}

func (np *NotificationPreference) CanSendPush() bool {
    return np.PushEnabled
}

func (np *NotificationPreference) CanSendEmail() bool {
    return np.EmailFrequency != "NEVER"
}

// NotificationLog represents a single delivery attempt
type NotificationLog struct {
    LogID            string    // UUID
    NotificationID   string
    Channel          DeliveryChannel
    DeliveryStatus   string    // SENT, FAILED, BOUNCED
    ResponseCode     int
    ErrorMessage     string
    RetryCount       int
    DeliveredAt      *time.Time
}

// Value Objects

type NotificationType string

const (
    MessageReceived NotificationType = "MESSAGE_RECEIVED"
    GroupCreated    NotificationType = "GROUP_CREATED"
    MemoryShared    NotificationType = "MEMORY_SHARED"
    MemoryLiked     NotificationType = "MEMORY_LIKED"
    CommentAdded    NotificationType = "COMMENT_ADDED"
    FriendAdded     NotificationType = "FRIEND_ADDED"
    UserMentioned   NotificationType = "USER_MENTIONED"
)

type DeliveryChannel string

const (
    ChannelPush   DeliveryChannel = "PUSH"
    ChannelEmail  DeliveryChannel = "EMAIL"
    ChannelInApp  DeliveryChannel = "IN_APP"
)

type NotificationPriority string

const (
    PriorityHigh   NotificationPriority = "HIGH"
    PriorityNormal NotificationPriority = "NORMAL"
    PriorityLow    NotificationPriority = "LOW"
)

type DevicePlatform string

const (
    PlatformiOS     DevicePlatform = "iOS"
    PlatformAndroid DevicePlatform = "Android"
    PlatformWeb     DevicePlatform = "Web"
)

type QuietHours struct {
    StartHour int    // 0-23
    EndHour   int    // 0-23
    Timezone  string
}

func (qh *QuietHours) IsInQuietHours(now time.Time) bool {
    hour := now.Hour()
    if qh.StartHour < qh.EndHour {
        return hour >= qh.StartHour && hour < qh.EndHour
    }
    // Spans midnight
    return hour >= qh.StartHour || hour < qh.EndHour
}

// Application layer - Use Cases

package usecase

type SendNotificationInput struct {
    UserID               string
    NotificationType     NotificationType
    Title                string
    Body                 string
    Data                 map[string]string
    Priority             NotificationPriority
}

type SendNotificationOutput struct {
    NotificationID       string
    ChannelsQueued       []DeliveryChannel
}

type SendNotificationUseCase struct {
    notificationRepo     NotificationRepository
    deviceTokenRepo      DeviceTokenRepository
    preferenceRepo       PreferenceRepository
    pushGateway          PushGateway        // FCMPushAdapter
    emailGateway         MailPort           // PostfixSMTPAdapter (Postfix + Dovecot + Rspamd)
    dedupeCache          Cache              // Redis (CachePort)
    eventPublisher       EventPublisher
}

func (uc *SendNotificationUseCase) Execute(ctx context.Context, in SendNotificationInput) (*SendNotificationOutput, error) {
    // 1. 重複排除チェック (Redis)
    dedupeKey := fmt.Sprintf("notif:dedup:%s:%s", in.UserID, in.NotificationType)
    if uc.dedupeCache.Get(dedupeKey) != nil {
        return nil, errors.New("duplicate notification within 24h")
    }

    // 2. ユーザーの通知設定を取得
    pref, err := uc.preferenceRepo.FindByUserAndType(ctx, in.UserID, in.NotificationType)
    if err != nil {
        return nil, err
    }

    // 3. 配信チャネルを決定
    channels := uc.determineChannels(pref)
    if len(channels) == 0 {
        return nil, errors.New("user has disabled all notification channels")
    }

    // 4. Notification エンティティを作成
    notification := &Notification{
        NotificationID:   uuid.New().String(),
        UserID:           in.UserID,
        Title:            in.Title,
        Body:             in.Body,
        NotificationType: in.NotificationType,
        DeliveryChannels: channels,
        Priority:         in.Priority,
        Status:           "PENDING",
        CreatedAt:        time.Now(),
        ExpiresAt:        time.Now().AddDate(0, 0, 7), // 7 days
        Data:             in.Data,
    }

    // 5. DB に保存
    if err := uc.notificationRepo.Save(ctx, notification); err != nil {
        return nil, err
    }

    // 6. 各チャネルで配信処理を開始
    for _, ch := range channels {
        switch ch {
        case "PUSH":
            go uc.sendPush(ctx, notification)
        case "EMAIL":
            go uc.sendEmail(ctx, notification)
        case "IN_APP":
            go uc.saveInApp(ctx, notification)
        }
    }

    // 7. 重複排除キャッシュに記録
    uc.dedupeCache.Set(dedupeKey, "1", 24*time.Hour)

    // 8. イベント発行
    uc.eventPublisher.PublishNotificationCreated(notification)

    return &SendNotificationOutput{
        NotificationID: notification.NotificationID,
        ChannelsQueued: channels,
    }, nil
}

type RegisterDeviceTokenInput struct {
    UserID   string
    DeviceID string
    Platform DevicePlatform
    Token    string
}

type RegisterDeviceTokenOutput struct {
    DeviceTokenID string
    ExpiresAt     time.Time
}

type RegisterDeviceTokenUseCase struct {
    deviceTokenRepo DeviceTokenRepository
    cache           Cache
    eventPublisher  EventPublisher
}

func (uc *RegisterDeviceTokenUseCase) Execute(ctx context.Context, in RegisterDeviceTokenInput) (*RegisterDeviceTokenOutput, error) {
    // 1. 既存トークンを検索
    existing, err := uc.deviceTokenRepo.FindByToken(ctx, in.Token)
    if err == nil && existing != nil {
        // 既存: is_valid を true に、last_used_at を更新
        existing.IsValid = true
        existing.LastUsedAt = time.Now()
        if err := uc.deviceTokenRepo.Update(ctx, existing); err != nil {
            return nil, err
        }
        return &RegisterDeviceTokenOutput{
            DeviceTokenID: existing.DeviceTokenID,
            ExpiresAt:     existing.ExpiresAt,
        }, nil
    }

    // 2. 新規登録
    expiresAt := time.Now().AddDate(1, 0, 0) // 1年有効
    dt := &DeviceToken{
        DeviceTokenID: uuid.New().String(),
        UserID:        in.UserID,
        DeviceID:      in.DeviceID,
        Platform:      in.Platform,
        TokenValue:    in.Token,
        IsValid:       true,
        RegisteredAt:  time.Now(),
        ExpiresAt:     expiresAt,
    }
    if err := uc.deviceTokenRepo.Save(ctx, dt); err != nil {
        return nil, err
    }

    // 3. キャッシュに記録
    uc.cache.Set(fmt.Sprintf("device_token:%s", in.Token), "1", 1*time.Hour)

    // 4. イベント発行
    uc.eventPublisher.PublishDeviceTokenRegistered(dt)

    return &RegisterDeviceTokenOutput{
        DeviceTokenID: dt.DeviceTokenID,
        ExpiresAt:     dt.ExpiresAt,
    }, nil
}

// Ports (Application layer interfaces)

// OutboundPorts (データ出力)

type NotificationRepository interface {
    Save(ctx context.Context, notif *Notification) error
    FindByID(ctx context.Context, id string) (*Notification, error)
    UpdateStatus(ctx context.Context, id string, status string) error
    ListByUser(ctx context.Context, userID string, limit int, offset int) ([]*Notification, int, error)
}

type DeviceTokenRepository interface {
    Save(ctx context.Context, dt *DeviceToken) error
    Update(ctx context.Context, dt *DeviceToken) error
    FindByToken(ctx context.Context, token string) (*DeviceToken, error)
    FindByUserAndPlatform(ctx context.Context, userID string, platform DevicePlatform) ([]*DeviceToken, error)
    Delete(ctx context.Context, id string) error
    ListValidByUser(ctx context.Context, userID string) ([]*DeviceToken, error)
}

type PreferenceRepository interface {
    Save(ctx context.Context, pref *NotificationPreference) error
    Update(ctx context.Context, pref *NotificationPreference) error
    FindByUserAndType(ctx context.Context, userID string, notifType NotificationType) (*NotificationPreference, error)
    FindByUser(ctx context.Context, userID string) ([]*NotificationPreference, error)
}

type NotificationLogRepository interface {
    Save(ctx context.Context, log *NotificationLog) error
    FindByNotificationID(ctx context.Context, notifID string) ([]*NotificationLog, error)
}

// Gateway interfaces for external services

type PushGateway interface {
    SendPush(ctx context.Context, notification *Notification, deviceTokens []*DeviceToken) ([]string, error) // Returns token IDs sent
}

// MailPort (Postfix SMTP adapter を実装)
type MailPort interface {
    SendEmail(ctx context.Context, notification *Notification, toAddress string) error
}

type EventConsumer interface {
    ConsumeMessage(ctx context.Context, message interface{}) error
}

type EventPublisher interface {
    PublishNotificationCreated(notif *Notification) error
    PublishNotificationSent(notif *Notification, channel DeliveryChannel) error
    PublishNotificationFailed(notif *Notification, channel DeliveryChannel, reason string) error
    PublishDeviceTokenRevoked(dt *DeviceToken, reason string) error
    PublishPreferencesUpdated(pref *NotificationPreference) error
}

type Cache interface {
    Get(key string) (interface{}, error)
    Set(key string, value interface{}, ttl time.Duration) error
    Delete(key string) error
}

// Adapters - External Services

package adapter

import (
    "context"
    firebase "firebase.google.com/go/v4"
    "firebase.google.com/go/v4/messaging"
)

type FCMPushAdapter struct {
    client *messaging.Client
}

func NewFCMPushAdapter(ctx context.Context, credentialsPath string) (*FCMPushAdapter, error) {
    opt := option.WithCredentialsFile(credentialsPath)
    app, err := firebase.NewApp(ctx, nil, opt)
    if err != nil {
        return nil, err
    }
    client, err := app.Messaging(ctx)
    if err != nil {
        return nil, err
    }
    return &FCMPushAdapter{client: client}, nil
}

func (fa *FCMPushAdapter) SendPush(ctx context.Context, notification *domain.Notification, deviceTokens []*domain.DeviceToken) ([]string, error) {
    var sentTokens []string
    for _, dt := range deviceTokens {
        if !dt.IsValid {
            continue
        }
        msg := &messaging.Message{
            Notification: &messaging.Notification{
                Title: notification.Title,
                Body:  notification.Body,
            },
            Data: notification.Data,
            Token: dt.TokenValue,
        }
        resp, err := fa.client.Send(ctx, msg)
        if err != nil {
            // Log failure, may add to retry queue
            continue
        }
        sentTokens = append(sentTokens, resp)
    }
    return sentTokens, nil
}

package adapter

import (
    "context"
    "crypto/tls"
    "errors"
    "fmt"
    "mime"
    "net/mail"
    "net/smtp"
    "strings"
)

type PostfixSMTPAdapter struct {
    host     string // e.g. "smtp.example.internal"
    port     int    // 587 (submission, STARTTLS)
    username string
    password string // SMTP auth token (password field)
    from     string // e.g. "no-reply@recuerdo.example"
}

func NewPostfixSMTPAdapter(cfg SMTPConfig) (*PostfixSMTPAdapter, error) {
    // 設定整合性の早期検証: 片方だけ設定された誤設定を起動時に検出する
    switch {
    case cfg.Username == "" && cfg.Password == "":
        // AUTH 無しモード（Postfix 側が submission relay 制限で運用）
    case cfg.Username != "" && cfg.Password != "":
        // 通常の AUTH 有りモード
    default:
        return nil, errors.New("SMTP username and token(password) must be set together (or both empty)")
    }
    if _, err := mail.ParseAddress(cfg.From); err != nil {
        return nil, fmt.Errorf("invalid SMTP From address: %w", err)
    }
    return &PostfixSMTPAdapter{
        host: cfg.Host, port: cfg.Port,
        username: cfg.Username, password: cfg.Password, from: cfg.From,
    }, nil
}

// sanitizeHeaderValue はヘッダ値から CR / LF を取り除き、ヘッダインジェクション
// （BCC 追加・追加ヘッダ注入）を防ぐ。値に混入していれば設定エラーとして拒否する。
func sanitizeHeaderValue(name, value string) (string, error) {
    if strings.ContainsAny(value, "\r\n") {
        return "", fmt.Errorf("header %q contains CR/LF", name)
    }
    return value, nil
}

func (a *PostfixSMTPAdapter) SendEmail(ctx context.Context, notif *domain.Notification, toAddress string) error {
    _ = ctx // net/smtp は context 非対応

    // 宛先・件名は必ず検証・エンコードしてからヘッダに組み込む
    to, err := mail.ParseAddress(toAddress)
    if err != nil {
        return err
    }
    defer c.Close()

    ok, _ := c.Extension("STARTTLS")
    if !ok {
        return fmt.Errorf("smtp server %s does not support STARTTLS", addr)
    }

    tlsConfig := &tls.Config{
        ServerName:         a.host,
        MinVersion:         tls.VersionTLS12,
        InsecureSkipVerify: false,
    }
    if err := c.StartTLS(tlsConfig); err != nil {
        return err
    }

    hasUsername := a.username != ""
    hasPassword := a.password != ""
    switch {
    case hasUsername && hasPassword:
        auth := smtp.PlainAuth("", a.username, a.password, a.host)
        ok, _ := c.Extension("AUTH")
        if !ok {
            return fmt.Errorf("smtp server %s does not support AUTH", addr)
        }
        if err := c.Auth(auth); err != nil {
            return err
        }
    case !hasUsername && !hasPassword:
        // AUTH 無しモード
    default:
        return errors.New("smtp auth misconfigured: username and token(password) must be set together")
    }

    if err := c.Mail(a.from); err != nil {
        return err
    }
    if err := c.Rcpt(toAddress); err != nil {
        return err
    }

    w, err := c.Data()
    if err != nil {
        return err
    }
    toHeader, err := sanitizeHeaderValue("To", to.Address)
    if err != nil {
        return err
    }
    // 非 ASCII 件名は RFC 2047（MIME encoded-word）でエンコード
    subject := mime.QEncoding.Encode("utf-8", notif.Title)
    if _, err := sanitizeHeaderValue("Subject", subject); err != nil {
        return err
    }

    var b strings.Builder
    fmt.Fprintf(&b, "From: %s\r\n", fromHeader)
    fmt.Fprintf(&b, "To: %s\r\n", toHeader)
    fmt.Fprintf(&b, "Subject: %s\r\n", subject)
    b.WriteString("MIME-Version: 1.0\r\n")
    b.WriteString("Content-Type: text/plain; charset=UTF-8\r\n")
    b.WriteString("Content-Transfer-Encoding: 8bit\r\n")
    b.WriteString("\r\n")
    b.WriteString(notif.Body)
    b.WriteString("\r\n")
    msg := []byte(b.String())

    return nil
}

package adapter

import (
    "context"
    "encoding/json"

    "github.com/hibiken/asynq"
    ociqueue "github.com/oracle/oci-go-sdk/v65/queue"
)

// ---- Beta: Redis + asynq 実装 ----
type RedisQueueConsumerAdapter struct {
    server           *asynq.Server
    sendNotification usecase.SendNotificationUseCase
}

func NewRedisQueueConsumerAdapter(redisAddr string, uc usecase.SendNotificationUseCase) *RedisQueueConsumerAdapter {
    srv := asynq.NewServer(
        asynq.RedisClientOpt{Addr: redisAddr},
        asynq.Config{Concurrency: 10, Queues: map[string]int{"notifications": 10}},
    )
    return &RedisQueueConsumerAdapter{server: srv, sendNotification: uc}
}

func (a *RedisQueueConsumerAdapter) Start(ctx context.Context) error {
    mux := asynq.NewServeMux()
    mux.HandleFunc("notification:send", func(ctx context.Context, t *asynq.Task) error {
        var in usecase.SendNotificationInput
        if err := json.Unmarshal(t.Payload(), &in); err != nil {
            return err
        }
        _, err := a.sendNotification.Execute(ctx, in)
        return err
    })
    return a.server.Run(mux)
}

// ---- Prod: OCI Queue 実装 ----
type OCIQueueConsumerAdapter struct {
    client           ociqueue.QueueClient
    queueID          string
    sendNotification usecase.SendNotificationUseCase
}

func (a *OCIQueueConsumerAdapter) ConsumeMessages(ctx context.Context) {
    for {
        resp, err := a.client.GetMessages(ctx, ociqueue.GetMessagesRequest{
            QueueId: &a.queueID,
        })
        if err != nil {
            continue
        }
        for _, msg := range resp.Messages {
            var in usecase.SendNotificationInput
            if err := json.Unmarshal([]byte(*msg.Content), &in); err != nil {
                continue
            }
            if _, err := a.sendNotification.Execute(ctx, in); err != nil {
                continue // DLQ に自動移動される
            }
            _, _ = a.client.DeleteMessage(ctx, ociqueue.DeleteMessageRequest{
                QueueId:   &a.queueID,
                MessageReceipt: msg.Receipt,
            })
        }
    }
}

package adapter

import (
    "context"
    "database/sql"
    "encoding/json"

    _ "github.com/go-sql-driver/mysql"
)

// MySQL 8.x / MariaDB 10.6+ 互換
type MySQLNotificationRepository struct {
    db *sql.DB
}

func (repo *MySQLNotificationRepository) Save(ctx context.Context, notif *domain.Notification) error {
    query := `INSERT INTO notifications (notification_id, user_id, title, body, notification_type, delivery_channels, priority, status, created_at, expires_at, data)
              VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`
    channelsJSON, _ := json.Marshal(notif.DeliveryChannels) // JSON 型カラム
    dataJSON, _ := json.Marshal(notif.Data)
    _, err := repo.db.ExecContext(ctx, query,
        notif.NotificationID, notif.UserID, notif.Title, notif.Body, notif.NotificationType,
        channelsJSON, notif.Priority, notif.Status, notif.CreatedAt, notif.ExpiresAt, dataJSON,
    )
    return err
}

// Unit test example: SendNotificationUseCase

func TestSendNotification_WithPushEnabled(t *testing.T) {
    // Arrange
    mockNotificationRepo := &MockNotificationRepository{}
    mockDeviceTokenRepo := &MockDeviceTokenRepository{}
    mockPushGateway := &MockPushGateway{}
    mockCache := &MockCache{}

    uc := &SendNotificationUseCase{
        notificationRepo: mockNotificationRepo,
        deviceTokenRepo: mockDeviceTokenRepo,
        pushGateway: mockPushGateway,
        dedupeCache: mockCache,
    }

    in := SendNotificationInput{
        UserID: "user123",
        NotificationType: "MESSAGE_RECEIVED",
        Title: "New message",
        Body: "Hello from Alice",
    }

    // Act
    out, err := uc.Execute(context.Background(), in)

    // Assert
    assert.NoError(t, err)
    assert.NotNil(t, out)
    assert.Equal(t, out.NotificationID, mockNotificationRepo.LastSaved.NotificationID)
    mockPushGateway.AssertSendPushCalled()
}