【Go×FCM】定期的にFCMへPush通知を送信する機能を実装：サーバー編

$ cd scripts/migrations
$ migrate create -ext sql -format 2006010215 add_fcm_token_to_users

// 2024050323_add_fcm_token_to_users.up.sql
ALTER TABLE users
    ADD COLUMN fcm_token VARCHAR(255),
    ADD COLUMN fcm_token_timestamp TIMESTAMP;

// 2024050323_add_fcm_token_to_users.down.sql
ALTER TABLE users
    DROP COLUMN fcm_token,
    DROP COLUMN fcm_token_timestamp;

// user.go
type User struct {
    UID                        string `db:"uid"`
    DeviceID                   string `db:"device_id"`
    Name                       string `db:"name"`
    // ... 他のフィールド
    FcmToken                   types.NullString `db:"fcm_token"`
    FcmTokenTimestamp          *time.Time      `db:"fcm_token_timestamp" json:"fcm_token_timestamp"`
}

// user_db.go
func UpdateUser(db *sqlx.DB, user *User) (*User, error) {
	tx, err := db.Beginx()
	if err != nil {
		return nil, err
	}

	query := `update users
			  set device_id				    = coalesce(:device_id, device_id),
			  ...
			  fcm_token = coalesce(:fcm_token, fcm_token),
			  fcm_token_timestamp		= coalesce(:fcm_token_timestamp, fcm_token_timestamp),
			  ...
			  WHERE uid = :uid;`
	return &updatedUser, nil
}

// lib/api/user.dart
class User {
  final String? uid;
  final String? fcmToken;
  final DateTime? fcmTokenTimestamp;

	User(
      {this.uid,
			 this.fcmToken,
       this.fcmTokenTimestamp});
}

flutter pub run build_runner build --delete-conflicting-outputs

import 'package:clocky_app/api/user.dart';
import 'package:clocky_app/api/user_notifier.dart';
import 'package:firebase_core/firebase_core.dart';
import 'package:firebase_messaging/firebase_messaging.dart';
import 'package:flutter/material.dart';
import 'package:flutter_riverpod/flutter_riverpod.dart';
import 'package:overlay_support/overlay_support.dart';

void main() async {
  WidgetsFlutterBinding.ensureInitialized();
  await Firebase.initializeApp(options: DefaultFirebaseOptions.currentPlatform);
  runApp(ProviderScope(child: const ClockyApp()));
}

class ClockyApp extends ConsumerStatefulWidget {
  const ClockyApp({super.key});

  @override
  _ClockyAppState createState() => _ClockyAppState();
}

class _ClockyAppState extends ConsumerState<ClockyApp> {
  @override
  void initState() {
    super.initState();
    initializeApp();
  }

  Future<void> initializeApp() async {
    final messaging = FirebaseMessaging.instance;
    NotificationSettings settings = await messaging.requestPermission(
      alert: true,
      announcement: false,
      badge: true,
      carPlay: false,
      criticalAlert: false,
      provisional: false,
      sound: true,
    );
    debugPrint('User granted permission: ${settings.authorizationStatus}');

    final token = await messaging.getToken();
    debugPrint('FCM TOKEN: $token');

    if (token != null) {
    // FCMトークンをサーバーに送る
      final userNotifier = ref.read(userProvider.notifier);
      userNotifier.updateUser(User(fcmToken: token, fcmTokenTimestamp: DateTime.now()));
    }
  }
}

flutter: Request updateUser: {"fcm_token":"xxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"}

POST https://fcm.googleapis.com/v1/projects/myproject-b5ae1/messages:send

  { "notification": {
      "body": "This is an FCM notification message!",
      "time": "FCM Message"
    },
    "to" : "bk3RNwTe3H0:CI2k_HHwgIpoDKCIZvvDMExUdFQ3P1..."
  }

package clocky_be

import (
	"context"
	"fmt"
	"log"

	fcm "google.golang.org/api/fcm/v1"
	"google.golang.org/api/option"
)


func SendFCMNotification(ctx context.Context, config *Config,token, title, body string) error {
	opt := option.WithCredentialsJSON([]byte(config.GoogleApplicationCredentials))
	service, err := fcm.NewService(ctx, opt)
	if err != nil {
		return fmt.Errorf("could not create FCM service: %v", err)
	}

	// Construct the message to be sent.
	message := &fcm.Message{
		Token: token,
		Notification: &fcm.Notification{
			Title: title,
			Body:  body,
		},
	}

	// Create a new "messages" client from the FCM service.
	messagesClient := fcm.NewProjectsMessagesService(service)

	request := &fcm.SendMessageRequest{
		Message: message,
	}
	fmt.Printf("FCM log")

	// Send the message using the FCM v1 API endpoint
	response, err := messagesClient.Send(fmt.Sprintf("projects/%s", config.FirebaseProjectId), request).Do()
	if err != nil {
		log.Printf("Request failed: %+v", request)
		return fmt.Errorf("failed to send notification: %v", err)
	}

	fmt.Printf("Successfully sent notification. Response: %+v\n", response)
	return nil
}

FCM_SERVER_KEY=your_fcm_server_key_here

// worker.go
// FCMを使用したリアルタイム天気通知の送信
func SendFCMWeatherNotification(ctx context.Context, db *sqlx.DB, message Message, config *Config) {
    user, err := GetUser(db, message.UserID)
    if err != nil {
        log.Printf("Failed to get user info for FCM notification: %v", err)
        return
    }

    // ユーザーのFCMトークンの存在を確認
    if !user.FcmToken.Valid {
        log.Printf("No FCM token available for user %s", user.UID)
        return
    }

    // 天気情報の更新が必要かどうかを確認
    if user.WeatherMessage.Valid && user.WeatherMessage.String != "" {
        // 天気情報をFCMで送信
        if err := SendFCMNotification(ctx, config, user.FcmToken.String, "[ミーア]今日の天気お知らせ", user.WeatherMessage.String); err != nil {
            log.Printf("Failed to send FCM notification for user %s: %v", user.UID, err)
            return
        }
        log.Printf("FCM notification sent for user: %s", user.UID)
    } else {
        log.Printf("No valid weather message to send for user %s", user.UID)
    }
}

// worker.go
func ScheduleTask(db *sqlx.DB, messagesChan chan Message) {
		var users []User
		query := `SELECT * FROM users WHERE DATE_FORMAT(weather_announcement_time, '%H:%i') = DATE_FORMAT(?, '%H:%i');`
		err = db.Select(&users, query, currentTimeJST.String())
    for _, user := range users {
			if user.WeatherAnnouncementTime.Valid {
				log.Printf("Sending FCM weather notification for user: %s", user.UID)
				message := Message{
					UserID: user.UID,
					Type:   "weather_fcm",
				}
				messagesChan <- message
			}
    }
    // ...
}

// worker.go
// メッセージ処理の抽象化
func ProcessMessages(db *sqlx.DB, messagesChan chan Message, config *Config) {
    for message := range messagesChan {
        log.Printf("Processing message for user: %s, type: %s", message.UserID, message.Type)

        // タイムアウトを設定
        ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
        defer cancel()

        switch message.Type {
        case "weather_fcm":
            SendFCMWeatherNotification(ctx, db, message, config)
        default:
            log.Printf("Unknown message type: %s", message.Type)
        }
    }
}

// main.go
package main

import (
	"context"
	"log"
	"github.com/jmoiron/sqlx"
	"github.com/robfig/cron/v3"
	"github.com/EarEEG-dev/clocky_be"
)

func main() {
	ctx := context.Background()
	config := clocky.NewConfig()

	db, err := sqlx.Open("mysql", config.DSN)
	if err != nil {
		log.Fatalf("unable to connect to database: %v", err)
	}

	if err := db.Ping(); err != nil {
		log.Fatalf("unable to ping database: %v", err)
	}
	log.Printf("connected to database")

	// メッセージチャネルとCronジョブの設定
	messagesChan := make(chan clocky.Message)

	// 秒単位をサポートしてcronを初期化
	c := cron.New(cron.WithSeconds())

	// 毎分0秒時点で起動
	c.AddFunc("0 * * * * *", func() {
		log.Println("Cron job triggered")
		clocky.ScheduleTask(db, messagesChan)
	})
	go c.Start()

	// 処理関数の起動数を増やして並列に処理させることも可能
	go clocky.ProcessMessages(db, messagesChan, config)

	// その他のアプリケーション初期化コード（HTTPサーバー、SQSリスナー、gRPCサーバー、Rest API）
	// ...
}