채팅 키워드알림 서비스 - 멀티테넌트 키워드 알림

// Matcher checks a message for configured keywords.
//
// Phase 2 note: multi-tenant support will require a richer return type
// that carries tenant identity alongside matched keywords. The consumer
// loop calls Match() and passes results to Notifier.Notify().
// Both interfaces are the swap points for Phase 2 without changing the loop itself.
type Matcher interface {
	Match(message string) []string
}

// 채팅 수집 서비스 워커: 스트림 퍼블리셔
const (
	firehoseKey    = "chat:firehose"
	channelMaxLen  = 1000
	firehoseMaxLen = 100000
)

// internal/consumer/consumer.go
const streamKey = "chat:firehose"

func (c *Consumer) Run(ctx context.Context) error {
	err := c.client.XGroupCreateMkStream(ctx, streamKey, c.group, "$").Err()
	if err != nil && !strings.Contains(err.Error(), "BUSYGROUP") {
		return err
	}

	go c.logStats(ctx)

	for {
		select {
		case <-ctx.Done():
			return nil
		default:
		}

		streams, err := c.client.XReadGroup(ctx, &redis.XReadGroupArgs{
			Group:    c.group,
			Consumer: c.name,
			Streams:  []string{streamKey, ">"},
			Count:    10,
			Block:    c.blockDuration, // 0 = infinite wait in prod; short poll in tests
		}).Result()
		...
	}
}

// MultiTenantMatcher holds a per-org KeywordMatcher and can match a message
// against all orgs in one pass.
type MultiTenantMatcher struct {
	orgMatchers map[int32]*KeywordMatcher
}

// MatchAll returns a map of orgID -> matched keywords for every org that
// has at least one keyword hit. Orgs with no matches are omitted.
func (m *MultiTenantMatcher) MatchAll(message string) map[int32][]string {
	results := make(map[int32][]string)
	for orgID, matcher := range m.orgMatchers {
		matched := matcher.Match(message)
		if len(matched) > 0 {
			results[orgID] = matched
		}
	}
	return results
}

func (c *Consumer) processMessage(ctx context.Context, msg redis.XMessage) {
	...
	configs := c.loader.GetConfigs()

	orgKeywords := make(map[int32][]string, len(configs))
	for orgID, cfg := range configs {
		if len(cfg.Keywords) > 0 && len(cfg.Groups) == 0 {
			orgKeywords[orgID] = cfg.Keywords
		}
	}
	...
	mtMatcher := matcher.NewMultiTenantMatcher(orgKeywords)

type keywordPattern struct {
	label string
	lower string
	regex *regexp.Regexp
}

var compiledKeywordCache sync.Map

func compileKeyword(keyword string) keywordPattern {
	trimmed := strings.TrimSpace(keyword)
	if cached, ok := compiledKeywordCache.Load(trimmed); ok {
		return cached.(keywordPattern)
	}

	pattern, ok := parseRegexKeyword(trimmed)
	if ok {
		if re, err := regexp.Compile(pattern); err == nil {
			compiled := keywordPattern{label: trimmed, regex: re}
			compiledKeywordCache.Store(trimmed, compiled)
			return compiled
		}
	}
	compiled := keywordPattern{label: trimmed, lower: strings.ToLower(trimmed)}
	compiledKeywordCache.Store(trimmed, compiled)
	return compiled
}

func parseRegexKeyword(keyword string) (string, bool) {
	if strings.HasPrefix(keyword, "regex:") {
		pattern := strings.TrimPrefix(keyword, "regex:")
		return pattern, pattern != ""
	}

	if len(keyword) < 2 || keyword[0] != '/' {
		return "", false
	}
	closing := strings.LastIndex(keyword[1:], "/")
	if closing < 0 {
		return "", false
	}
	closing += 1
	pattern := keyword[1:closing]
	flags := keyword[closing+1:]
	if pattern == "" || !isSupportedSlashFlags(flags) {
		return "", false
	}
	if flags == "" {
		return pattern, true
	}
	return "(?" + uniqueFlags(flags) + ")" + pattern, true
}

func isSupportedSlashFlags(flags string) bool {
	for _, flag := range flags {
		if flag != 'i' && flag != 'm' && flag != 's' {
			return false
		}
	}
	return true
}

var compiledKeywordCache sync.Map

for _, kw := range g.Keywords {
	pattern := compileKeyword(kw)
	...
}

func (m *KeywordMatcher) Match(message string) []string {
	lower := strings.ToLower(message)
	var matched []string
	for _, p := range m.patterns {
		if p.regex != nil {
			if p.regex.MatchString(message) {
				matched = append(matched, p.label)
			}
			continue
		}
		if strings.Contains(lower, p.lower) {
			matched = append(matched, p.label)
		}
	}
	return matched
}

if p.regex.MatchString(message)      // 원본
if strings.Contains(lower, p.lower)  // 소문자

func (m *MultiTenantMatcher) MatchGroups(
	orgID int32,
	groups []model.GroupConfig,
	message string,
) []GroupMatch {
	var out []GroupMatch
	lower := strings.ToLower(message)
	for _, g := range groups {
		var matched []string
		for _, kw := range g.Keywords {
			pattern := compileKeyword(kw)
			if pattern.regex != nil {
				if pattern.regex.MatchString(message) {
					matched = append(matched, pattern.label)
				}
				continue
			}
			if strings.Contains(lower, pattern.lower) {
				matched = append(matched, pattern.label)
			}
		}
		...
	}
}

func newTestMatcher(t *testing.T) *MultiTenantMatcher {
	t.Helper()
	return NewMultiTenantMatcher(map[int32][]string{})
}

// Platform comparison is case-insensitive. UserID is exact match (case-sensitive)
// because platform user IDs are opaque tokens.
func (m *MultiTenantMatcher) MatchUserGroups(
	orgID int32,
	groups []model.UserGroupConfig,
	platform string,
	userID string,
) []UserGroupMatch {
	if userID == "" {
		return nil
	}
	var out []UserGroupMatch
	for _, g := range groups {
		var matched []model.AlertUser
		for _, u := range g.Users {
			if strings.EqualFold(u.Platform, platform) && u.PlatformUserID == userID {
				matched = append(matched, u)
			}
		}
		...
	}
}

// internal/model/config.go

// ChannelConfig describes a single alert destination with full metadata.
type ChannelConfig struct {
	ChannelID  int32
	Name       string
	Type       string // "SLACK" | "DISCORD" | "PUSH"
	WebhookURL string
}

// GroupConfig holds the keyword-alert configuration for a single group
// within an organisation.
type GroupConfig struct {
	GroupID  int32
	Name     string
	IsActive bool
	Keywords []string
	Channels []ChannelConfig
}

// AlertUser identifies a specific viewer inside a UserGroupConfig.
type AlertUser struct {
	Platform       string
	PlatformUserID string
	DisplayLabel   string
}

type UserGroupConfig struct {
	GroupID  int32
	Name     string
	IsActive bool
	Users    []AlertUser
	Channels []ChannelConfig
}

// internal/configloader/loader.go

// AlertChannelConfig describes a single alert destination.
// Kept for backward compatibility with notifier.Router.SendAll.
type AlertChannelConfig struct {
	Type       string
	WebhookURL string
}

type OrgConfig struct {
	OrganizationID int32
	IsActive       bool
	Groups         []model.GroupConfig     // primary path (new): keyword-target alerts
	UserGroups     []model.UserGroupConfig // user-target alerts

	// legacy: populated by loader as fallback when Groups is empty
	Keywords []string
	Channels []AlertChannelConfig
}

func modelChannelToAlert(ch model.ChannelConfig) configloader.AlertChannelConfig {
	return configloader.AlertChannelConfig{
		Type:       ch.Type,
		WebhookURL: ch.WebhookURL,
	}
}

model KeywordAlertConfig {
  id             Int      @id @default(autoincrement())
  organizationId Int      @unique
  isActive       Boolean  @default(true)
  // legacy: kept nullable for safe rollback. dropped in cleanup PR.
  keywords       Json?

  organization Organization            @relation(...)
  channels     KeywordAlertChannel[]
  groups       KeywordAlertGroup[]
  userGroups   KeywordAlertUserGroup[]
}

model KeywordAlertChannel {
  id         Int              @id @default(autoincrement())
  configId   Int
  name       String?          @db.VarChar(100)
  type       AlertChannelType
  webhookUrl String?
  isActive   Boolean          @default(true)
}

model KeywordAlertGroup {
  id       Int     @id @default(autoincrement())
  configId Int
  name     String  @db.VarChar(100)
  keywords Json    @default("[]")
  isActive Boolean @default(true)

  channelLinks KeywordAlertGroupChannel[]
}

model KeywordAlertGroupChannel {
  id        Int @id @default(autoincrement())
  groupId   Int
  channelId Int

  @@unique([groupId, channelId])
  @@index([channelId])
}

model KeywordAlertUserGroupUser {
  id             Int    @id @default(autoincrement())
  groupId        Int
  platform       String @db.VarChar(50)
  platformUserId String @db.VarChar(200)
  displayLabel   String? @db.VarChar(200)

  @@unique([groupId, platform, platformUserId])
}

// 백엔드: findAllActiveConfigs()
return {
  organization_id: config.organizationId,
  is_active: config.isActive,
  keywords: legacyKeywords,
  channels: channelPool,     // legacy top-level mirror
  channel_pool: channelPool,
  groups,                    // groups[].channel_ids
  user_groups: userGroups,   // user_groups[].channel_ids
};

// resolveChannels picks the new (channel_ids + pool) path when populated,
// falling back to the deprecated embedded channels for transitional servers.
resolveChannels := func(ids []int32, embedded []*businessv1.AlertChannel) []model.ChannelConfig {
	if len(ids) > 0 {
		out := make([]model.ChannelConfig, 0, len(ids))
		for _, id := range ids {
			if ch, ok := channelPool[id]; ok {
				out = append(out, ch)
			}
		}
		return out
	}
	out := make([]model.ChannelConfig, 0, len(embedded))
	for _, ch := range embedded {
		out = append(out, model.ChannelConfig{
			ChannelID:  ch.GetChannelId(),
			Name:       ch.GetName(),
			Type:       ch.GetType(),
			WebhookURL: ch.GetWebhookUrl(),
		})
	}
	return out
}

function hasUnsupportedGoRegexSyntax(pattern: string): boolean {
  return (
    pattern.includes('(?=') ||
    pattern.includes('(?!') ||
    pattern.includes('(?<=') ||
    pattern.includes('(?<!') ||
    /(^|[^\\])\\[1-9]/.test(pattern)
  );
}

func (l *Loader) Run(ctx context.Context) {
	ticker := time.NewTicker(l.pollInterval)
	defer ticker.Stop()

	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			l.fetch(ctx)
		}
	}
}

// Per-call deadline prevents polling stalls when the backend is slow/hung.
callCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
defer cancel()

client := businessv1.NewKeywordAlertQueryServiceClient(conn)
resp, err := client.GetKeywordConfigs(callCtx, &businessv1.GetKeywordConfigsRequest{})

func (l *Loader) updateConfigs(configs []*OrgConfig) {
	m := make(map[int32]*OrgConfig, len(configs))
	for _, c := range configs {
		m[c.OrganizationID] = c
	}

	l.mu.Lock()
	l.configs = m
	l.mu.Unlock()
}

// GetConfigs returns a deep-copied snapshot of the current configs.
// Safe for concurrent use.
func (l *Loader) GetConfigs() map[int32]*OrgConfig {
	l.mu.RLock()
	defer l.mu.RUnlock()

	snapshot := make(map[int32]*OrgConfig, len(l.configs))
	for id, cfg := range l.configs {
		cp := &OrgConfig{...}
		copy(cp.Keywords, cfg.Keywords)
		copy(cp.Channels, cfg.Channels)
		// Groups, UserGroups도 슬라이스까지 복사
		...
	}
	return snapshot
}

// Require first successful config fetch before consuming. This prevents
// silent message loss when the backend gRPC is unreachable at startup.
// Retry every 2s up to 30s total before failing.
initCtx, initCancel := context.WithTimeout(ctx, 30*time.Second)
var initErr error
for {
	initErr = loader.FetchOnceWithError(initCtx)
	if initErr == nil {
		break
	}
	slog.Warn("initial config fetch failed, retrying", "err", initErr)
	select {
	case <-initCtx.Done():
		initCancel()
		log.Fatalf("config loader startup failed after 30s: %v", initErr)
	case <-time.After(2 * time.Second):
	}
}

// ClickHouse (opt-in via ADDR). When configured, fail startup on connection
// failure to avoid silently losing history for the whole process lifetime.
if cfg.ClickHouseAddr != "" {
	chWriter, err = store.NewClickHouseWriter(...)
	if err != nil {
		log.Fatalf("clickhouse connection failed (fail-closed): %v", err)
	}
}

// dedupeGroupMatches combines per-group channels into a single unique slice
// (keyed by ChannelID) and unions the matched keywords across groups.
func dedupeGroupMatches(matches []matcher.GroupMatch) (channels []model.ChannelConfig, keywords []string) {
	seen := make(map[int32]struct{})
	keywordSet := make(map[string]struct{})
	for _, gm := range matches {
		for _, ch := range gm.Channels {
			if _, ok := seen[ch.ChannelID]; ok {
				continue
			}
			seen[ch.ChannelID] = struct{}{}
			channels = append(channels, ch)
		}
		for _, kw := range gm.Keywords {
			keywordSet[kw] = struct{}{}
		}
	}
	for kw := range keywordSet {
		keywords = append(keywords, kw)
	}
	return
}

func dedupeUserGroupMatches(matches []matcher.UserGroupMatch) (channels []model.ChannelConfig, labels []string) {
	seen := make(map[int32]struct{})
	labelSet := make(map[string]struct{})
	for _, ugm := range matches {
		for _, ch := range ugm.Channels {
			...
		}
		labelSet[ugm.Name] = struct{}{}
	}
	...
}

// Per-org matching results: keyword and user paths tracked independently
// because alert content (matched keywords vs. user-group labels) differs.
type orgResult struct {
	labels   []string
	channels []configloader.AlertChannelConfig
}
orgKeywordResults := make(map[int32]orgResult, len(configs))
orgUserResults := make(map[int32]orgResult, len(configs))

// Sender sends a notification to a webhook-based destination.
type Sender interface {
	Send(webhookURL string, msg model.ChatMessage, keywords []string) error
}

// PushSender sends push notifications for a given organization.
type PushSender interface {
	SendPush(ctx context.Context, orgID int32, msg model.ChatMessage, keywords []string) error
}

// SendResult tracks success and failure counts across all channels.
type SendResult struct {
	Success int
	Failure int
}

// SendAll iterates the alert channels and dispatches each to the correct
// notifier. Errors are logged but not returned: a failure on one channel
// must not block delivery to the others. Returns counts for metrics.
func (r *Router) SendAll(ctx context.Context, orgID int32, channels []configloader.AlertChannelConfig, msg model.ChatMessage, keywords []string) SendResult {
	result := SendResult{}
	for _, ch := range channels {
		var err error
		switch strings.ToUpper(ch.Type) {
		case "SLACK":
			err = r.slack.Send(ch.WebhookURL, msg, keywords)
		case "DISCORD":
			err = r.discord.Send(ch.WebhookURL, msg, keywords)
		case "PUSH":
			err = r.push.SendPush(ctx, orgID, msg, keywords)
		default:
			log.Printf("[router] unknown channel type %q (org=%d)", ch.Type, orgID)
			continue
		}
		if err != nil {
			log.Printf("[router] %s send error (org=%d): %v", ch.Type, orgID, err)
			result.Failure++
		} else {
			result.Success++
		}
	}
	return result
}

channels := []configloader.AlertChannelConfig{
	{Type: "slack", WebhookURL: "https://example.test/hook-a"},  // lowercase
	{Type: "SLACK", WebhookURL: "https://example.test/hook-b"},  // uppercase
	{Type: "Slack", WebhookURL: "https://example.test/hook-c"},  // mixed
}
result := r.SendAll(context.Background(), 1, channels, msg, []string{"kw"})
// 3 sends, result.Success == 3

func buildPayload(msg model.ChatMessage, keywords []string) slackPayload {
	url := channelURL(msg.Platform, msg.ChannelID)
	label := fmt.Sprintf("[%s/%s]", escapeMrkdwn(msg.Platform), escapeMrkdwn(msg.StreamerName))
	keywordLabels := formatKeywordLabels(keywords)
	var headerText string
	if url != "" {
		headerText = fmt.Sprintf("*<%s|%s>*  `%s`", url, label, strings.Join(keywordLabels, ", "))
	} else {
		headerText = fmt.Sprintf("*%s*  `%s`", label, strings.Join(keywordLabels, ", "))
	}
	if msg.IsMelomingChannel {
		headerText = ":meloming_new: " + headerText
	}

	author := msg.Nickname
	if author == "" {
		author = "?"
	}
	bodyText := fmt.Sprintf("%s : %s", escapeMrkdwn(author), escapeMrkdwn(msg.Message))

	return slackPayload{
		Blocks: []slackBlock{
			{
				Type: "section",
				Text: &slackText{Type: "mrkdwn", Text: headerText + "\n" + bodyText},
			},
		},
	}
}

func channelURL(platform, channelID string) string {
	switch platform {
	case "chzzk":
		return "https://chzzk.naver.com/live/" + channelID
	case "soop":
		return "https://play.sooplive.com/" + channelID
	case "cime":
		return "https://ci.me/@" + channelID + "/live"
	default:
		return ""
	}
}

// escapeMrkdwn escapes Slack mrkdwn control characters in user-supplied strings.
// Prevents injection of <!channel>, <!here>, and crafted links.
func escapeMrkdwn(s string) string {
	s = strings.ReplaceAll(s, "&", "&amp;")
	s = strings.ReplaceAll(s, "<", "&lt;")
	s = strings.ReplaceAll(s, ">", "&gt;")
	return s
}

endpoint := fmt.Sprintf(
	"%s/v1/channel/by-platform/%s/%s",
	l.baseURL,
	url.PathEscape(platform),
	url.PathEscape(platformChannelID),
)

const (
	registeredChannelCacheTTL   = 5 * time.Minute
	unregisteredChannelCacheTTL = time.Minute
)

registered, err := n.channelLookup.IsRegistered(context.Background(), msg.Platform, msg.ChannelID)
if err != nil {
	log.Printf("[slack] meloming channel lookup failed (platform=%s channel_id=%s): %v", ...)
} else {
	msg.IsMelomingChannel = registered
}

type discordEmbed struct {
	Title       string `json:"title"`
	Description string `json:"description"`
	Color       int    `json:"color"`
	Timestamp   string `json:"timestamp"`
}

keywordLabels := formatKeywordLabels(keywords)
description := fmt.Sprintf("매칭 키워드: %s\n%s : %s",
	strings.Join(keywordLabels, ", "),
	nickname,
	msg.Message,
)

// Use the message timestamp if valid RFC3339, otherwise use current time.
ts := msg.Timestamp
if _, err := time.Parse(time.RFC3339, ts); err != nil {
	ts = time.Now().UTC().Format(time.RFC3339)
}

// Discord returns 204 No Content on success.
if resp.StatusCode >= 200 && resp.StatusCode < 300 {
	return nil
}

// GrpcPushClient sends push notifications via gRPC to the backend's
// KeywordAlertCommandService.SendPushNotification.
func (c *GrpcPushClient) SendPush(ctx context.Context, orgID int32, title, body, url string) error {
	conn, err := grpc.NewClient(c.addr, grpc.WithTransportCredentials(insecure.NewCredentials()))
	if err != nil {
		return fmt.Errorf("grpc dial: %w", err)
	}
	defer conn.Close()

	callCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
	defer cancel()

	client := businessv1.NewKeywordAlertCommandServiceClient(conn)
	_, err = client.SendPushNotification(callCtx, &businessv1.SendPushNotificationRequest{
		OrganizationId: orgID,
		Title:          title,
		Body:           body,
		Url:            url,
	})
	...
}

@GrpcMethod('KeywordAlertCommandService', 'SendPushNotification')
async sendPushNotification(data: {...}) {
  const members = await this.prisma.organizationMember.findMany({
    where: { organizationId: data.organization_id },
    select: { userId: true },
  });

  for (const member of members) {
    await this.notifications.enqueueSend(member.userId, {
      type: 'KEYWORD_ALERT',
      title: data.title,
      body: data.body,
      url: data.url,
    });
  }

  return {};
}

func formatKeywordLabel(keyword string) string {
	trimmed := strings.TrimSpace(keyword)
	if !isRegexKeyword(trimmed) {
		return keyword
	}
	return fmt.Sprintf("regex(%d)", keywordTraceID(trimmed))
}

func keywordTraceID(keyword string) uint32 {
	h := fnv.New32a()
	_, _ = h.Write([]byte(keyword))
	return h.Sum32()
}

func (c *Consumer) processMessage(ctx context.Context, msg redis.XMessage) {
	c.metrics.MessagesConsumed.Add(1)
	c.consumedCount.Add(1)

	// ACK with an independent context so SIGTERM cancellation does not skip the ACK.
	// A 5s deadline prevents a hung Redis connection from blocking shutdown indefinitely.
	// Registered first (before any early return) to avoid PEL accumulation on excluded senders.
	ackCtx, ackCancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer ackCancel()
	defer func() {
		if err := c.client.XAck(ackCtx, streamKey, c.group, msg.ID).Err(); err != nil {
			log.Printf("consumer: XACK failed for %s: %v", msg.ID, err)
		}
	}()

	chatMsg := parseMessage(msg.Values)

	if chatMsg.Type == "system" {
		return
	}
	if strings.EqualFold(chatMsg.Platform, "chzzk") && chatMsg.UserID == "SYSTEM_MESSAGE" {
		return
	}
	if chatMsg.Message == "" {
		return
	}
	if c.isExcludedSender(chatMsg.Platform, chatMsg.Nickname) {
		return
	}

func (c *Consumer) isExcludedSender(platform, nickname string) bool {
	platformLower := strings.ToLower(platform)
	nicknameLower := strings.ToLower(nickname)
	for _, ex := range c.excludedSenders {
		if strings.ToLower(ex.Platform) == platformLower && strings.ToLower(ex.Nickname) == nicknameLower {
			return true
		}
	}
	return false
}

// MatchRecord represents a single keyword match to be written to ClickHouse.
type MatchRecord struct {
	OrganizationID int32
	Keyword        string
	StreamID       string // Redis Stream message ID
	Msg            model.ChatMessage
	MatchedAt      time.Time
}

func (w *ClickHouseWriter) Write(ctx context.Context, records []MatchRecord) error {
	if len(records) == 0 {
		return nil
	}

	batch, err := w.conn.PrepareBatch(ctx, `INSERT INTO keyword_alert_matches
		(organization_id, keyword, stream_id, platform, channel_id, streamer_name,
		 user_id, nickname, message, matched_at)`)
	...
}

-- ClickHouse DDL
keyword_alert_matches (
    organization_id  UInt32,
    keyword          String,
    stream_id        String,                  -- Redis Stream message ID
    platform         LowCardinality(String),  -- "chzzk", "soop", "cime"
    channel_id       String,
    streamer_name    String,
    user_id          String,
    nickname         String,
    message          String,
    matched_at       DateTime('Asia/Seoul'),
    inserted_at      DateTime DEFAULT now()
) ENGINE = MergeTree()
PARTITION BY toYYYYMM(matched_at)
ORDER BY (organization_id, matched_at, stream_id)
TTL matched_at + INTERVAL 90 DAY;

for orgID, res := range orgKeywordResults {
	for _, kw := range res.labels {
		records = append(records, store.MatchRecord{
			OrganizationID: orgID,
			Keyword:        kw,
			StreamID:       msg.ID,
			Msg:            chatMsg,
			MatchedAt:      now,
		})
	}
}

if err := c.store.Write(ctx, records); err != nil {
	log.Printf("consumer: ClickHouse write error: %v", err)
}

const conditions = ['organization_id = {orgId:UInt32}'];
if (query.keyword)  conditions.push('keyword = {keyword:String}');
if (query.platform) conditions.push('platform = {platform:String}');
if (query.from)     conditions.push("matched_at >= {from:DateTime64(3, 'UTC')}");
if (query.to)       conditions.push("matched_at <= {to:DateTime64(3, 'UTC')}");

MessagesConsumed: prometheus.NewCounter(prometheus.CounterOpts{
	Name: "keyword_alert_messages_consumed_total",
	Help: "Total chat messages consumed from chat:firehose",
}),
Matches: prometheus.NewCounter(prometheus.CounterOpts{
	Name: "keyword_alert_matches_total",
	Help: "Total messages matching at least one keyword",
}),
SlackSent: prometheus.NewCounterVec(prometheus.CounterOpts{
	Name: "keyword_alert_slack_sent_total",
	Help: "Total Slack notifications attempted",
}, []string{"result"}),
SlackDuration: prometheus.NewHistogram(prometheus.HistogramOpts{
	Name:    "keyword_alert_slack_send_duration_seconds",
	Help:    "Slack webhook send duration in seconds",
	Buckets: prometheus.DefBuckets,
}),

start := time.Now()
totalSuccess, totalFailure := 0, 0
for orgID, res := range orgKeywordResults {
	result := c.router.SendAll(ctx, orgID, res.channels, chatMsg, res.labels)
	totalSuccess += result.Success
	totalFailure += result.Failure
}
for orgID, res := range orgUserResults {
	result := c.router.SendAll(ctx, orgID, res.channels, chatMsg, res.labels)
	...
}
c.metrics.SlackDuration.Observe(time.Since(start).Seconds())

if totalSuccess > 0 {
	c.metrics.SlackSent.WithLabelValues("success").Add(float64(totalSuccess))
}
if totalFailure > 0 {
	c.metrics.SlackSent.WithLabelValues("error").Add(float64(totalFailure))
}

healthSrv := &http.Server{Addr: ":8080", Handler: healthMux}   // /healthz
metricsSrv := &http.Server{Addr: ":9090", Handler: metricsMux} // /metrics

func (c *Consumer) logStats(ctx context.Context) {
	ticker := time.NewTicker(1 * time.Minute)
	...
	current := c.consumedCount.Load()
	log.Printf("stats: consumed %d messages in the last minute (total: %d)", current-last, current)
}

FROM golang:1.26-alpine AS builder
ENV GOPRIVATE=github.com/<org>/*
...
RUN --mount=type=secret,id=github_token \
    if [ -f /run/secrets/github_token ]; then \
      git config --global url."https://x-access-token:$(cat /run/secrets/github_token)@github.com/".insteadOf "https://github.com/"; \
    fi && \
    go mod download && ...

RUN CGO_ENABLED=0 GOOS=linux go build -o /keyword-alert-service ./cmd/

FROM gcr.io/distroless/static-debian12:nonroot
COPY --from=builder /keyword-alert-service /keyword-alert-service
USER nonroot:nonroot
ENTRYPOINT ["/keyword-alert-service"]

# .github/workflows/deploy-prod.yml (요약)
env:
  IMAGE_REPOSITORY: <사내 컨테이너 레지스트리>/keyword-alert-service

jobs:
  test:            # go test -race + build check
  build-and-push:  # runs-on: ubuntu-24.04-arm, --platform linux/arm64
  deploy:          # 배포 설정 저장소의 프로덕션 values 파일 태그 갱신 후 push

replicaCount: 1
revisionHistoryLimit: 1

strategy:
  type: Recreate

env:
  - name: CONSUMER_NAME
    valueFrom:
      fieldRef:
        fieldPath: metadata.name
  - name: EXCLUDED_SENDERS
    value: "chzzk/example-bot,soop/example-bot"
  - name: GRPC_ADDR
    value: "<백엔드 gRPC 주소>"
  - name: REDIS_ADDR
    value: "<채팅 버퍼 Redis 주소>:6379"
  - name: CLICKHOUSE_ADDR
    value: "<채팅 버퍼 ClickHouse 주소>:9000"

resources:
  # Right-sized from 30d peak: mem 16Mi, cpu small.
  requests: { cpu: 50m, memory: 64Mi }
  limits:   { cpu: 300m, memory: 192Mi }

service:     { enabled: false }
ingress:     { enabled: false }
autoscaling: { enabled: false }

nodeSelector:
  kubernetes.io/arch: arm64

tolerations:
  - key: node.kubernetes.io/lifecycle
    operator: Equal
    value: spot
    effect: NoSchedule

secrets:
  enabled: true
  refreshInterval: 1m
  keys:
    - SLACK_WEBHOOK_URL
    - CONSUMER_GROUP
    - CLICKHOUSE_PASSWORD

metrics:
  enabled: true
  port: 9090
  path: /metrics
  scrape:
    enabled: true
    interval: 30s