Import devices from OneSignal into Customer.io and convert APNS tokens to FCM.

import_one_signal_into_customerio.dart

/// Script used to import user devices (notification tokens) from
/// OneSignal into Customer.io. In the process, it converts APNS
/// tokens to FCM tokens.
///
/// See:
/// - https://www.thepolyglotdeveloper.com/2017/06/apns-tokens-fcm-tokens-simple-http
/// - https://developers.google.com/instance-id/reference/server#create_registration_tokens_for_apns_tokens
/// - https://www.customer.io/docs/api/#operation/add_device

import 'dart:convert';
import 'dart:io';

import 'package:collection/collection.dart';
import 'package:csv/csv.dart';
import 'package:http/http.dart' as http;

const _kChannelAndroid = 'android';
const _kChannelIOS = 'ios';

/// See: https://www.customer.io/docs/api/#section/Authentication/Basic-Auth-(Tracking-API-Key)
const _kAuthCustomerIo = '...';

/// Firebase Console > Project Settings > Cloud Messaging > Cloud Messaging API (Legacy) > Server Key.
const _kFirebaseServerKey = '...';

/// The packange name / bundle identifier of your app.
const _kBundleId = 'com.example.app';

/// CSV export obtained from OneSignal.
///
/// See: https://documentation.onesignal.com/docs/user-data.
final source = File('one_signal_export.csv');

int getColumnIndex(List<dynamic> columns, String columnName) {
  final index = columns.indexOf(columnName);
  if (index == -1) {
    throw 'Column $columnName not found';
  }
  return index;
}

void main() async {
  final users = readUsers();

  final androidUsers =
      users.where((user) => user['channel'] == _kChannelAndroid);
  final iosUsers = users.where((it) => it['channel'] == _kChannelIOS).toList();
  final iosUsersWithConvertedTokens = await convertToFcmTokens(iosUsers);

  print('Combining batches...');
  final finalUsers = [...androidUsers, ...iosUsersWithConvertedTokens];
  print('Combining batches OK.');

  // You can use this file to import your users to customer.io, 
  // or make sure they exist.
  writeUsers(users, 'filtered_users');

  print('Exporting users CSV with converted tokens...');
  writeUsers(finalUsers, 'users_with_converted_tokens.csv');
  print('Exporting OK.');

  await importIntoCustomerIo(finalUsers);

  print('Done.');
}

List<Map<String, dynamic>> readUsers() {
  print('Reading file...');
  final rows = CsvToListConverter().convert(
    source.readAsStringSync(),
    eol: '\n',
  );
  print('Reading OK.');

  final names = rows.firstOrNull;
  if (names == null) {
    throw 'Column names not found.';
  }

  print('Finding column indices...');
  final channelIndex = getColumnIndex(names, 'channel');
  final externalUserIdIndex = getColumnIndex(names, 'external_user_id');
  final pushTokenIndex = getColumnIndex(names, 'push_token');
  print('Finding column indices OK.');

  print('Filtering iOS and Android users and checking for data validity...');
  final users = rows
      .where((row) {
        row[channelIndex] = row[channelIndex]
            ?.toString()
            .toLowerCase()
            .replaceAll('google', '');
        final isAndroid = row[channelIndex] == _kChannelAndroid;
        final isIOS = row[channelIndex] == _kChannelIOS;
        final hasToken =
            row[pushTokenIndex]?.toString().trim().isNotEmpty == true;
        final hasId =
            row[externalUserIdIndex]?.toString().trim().isNotEmpty == true;

        return (isAndroid || isIOS) && hasId && hasToken;
      })
      .map((row) => {
            'id': row[externalUserIdIndex],
            'channel': row[channelIndex],
            'token': row[pushTokenIndex],
          })
      .toList();
  print('Filtering and checking OK.');

  return users;
}

void writeUsers(List<Map<String, dynamic>> users, String filename) {
  final csvList = [
    ['id', 'channel', 'token'],
    ...users.map((e) => [e['id'], e['channel'], e['token']]),
  ];
  File('$filename.csv').writeAsStringSync(
    const ListToCsvConverter().convert(csvList),
  );
}

Future<List<Map<String, dynamic>>> convertToFcmTokens(
    List<Map<String, dynamic>> users) async {
  final tokens = users.map((user) => user['token'].toString()).toList();

  print(
    'Converting ${tokens.length} APNS tokens. '
    'Sample: ${tokens.take(2).join(', ')}.',
  );

  final batches = tokens.slices(100); // Max 100 tokens allowed per request.

  final List<Map<String, dynamic>> processedUsers = [];

  print('Making HTTP requests to Google to get FCM tokens...');
  for (int i = 0; i < batches.length; i++) {
    final batch = batches.elementAt(i);
    print('Batch ${i + 1}/${batches.length}...');
    final response = await http.post(
      Uri.parse('https://iid.googleapis.com/iid/v1:batchImport'),
      headers: {
        'Authorization': 'key=$_kFirebaseServerKey',
        'Content-Type': 'application/json',
      },
      body: json.encode({
        'application': '$_kBundleId',
        'sandbox': false,
        'apns_tokens': batch,
      }),
    );

    if (response.statusCode == 200) {
      print('\tHTTP request OK.');

      print('\tDecoding response...');
      final results = json.decode(response.body)['results'] as List;
      print('\tDecoding response OK.');

      final okResults = results.where((it) => it['status'] == 'OK').toList();
      final nonOkResults = results.where((it) => it['status'] != 'OK').toList();

      print('\tCombining new FCM tokens with existing user data...');
      final processed = okResults.map((it) {
        final user =
            users.firstWhere((user) => user['token'] == it['apns_token']);
        return {
          ...user,
          'token': it['registration_token'],
        };
      }).toList();
      print('\tCombining OK.');

      print(
        '\tFailed to convert ${nonOkResults.length} tokens. '
        'See failed_tokens.json.',
      );
      File('failed_tokens.json').writeAsStringSync(json.encode(nonOkResults));

      processedUsers.addAll(processed);
    } else {
      throw 'HTTP request failed when converting APNS tokens to FCM tokens. ${response.reasonPhrase} - ${response.body}';
    }
  }

  return processedUsers;
}

Future importIntoCustomerIo(List<Map<String, dynamic>> users) async {
  print('Registering devices to Customer.io users...');

  List<Map<String, dynamic>> failedUsers = [];
  List<Map<String, dynamic>> importedUsers = [];

  for (final user in users) {
    final response = await http.put(
      Uri.parse(
          'https://track.customer.io/api/v1/customers/${user['id']}/devices'),
      headers: {
        'Authorization': 'Basic $_kAuthCustomerIo',
        'Content-Type': 'application/json',
      },
      body: json.encode({
        'device': {
          'id': user['token'],
          'platform': user['channel']?.toString().toLowerCase(),
        },
      }),
    );
    if (response.statusCode == 200) {
      importedUsers.add(user);
    } else {
      print(response.body);
      failedUsers.add(
        {
          ...user,
          'reason':
              '${response.statusCode} - ${response.headers} - ${response.body}'
        },
      );
    }

    File('customerio_failed_users.json')
        .writeAsStringSync(json.encode(failedUsers));
    File('customerio_imported_users.json')
        .writeAsStringSync(json.encode(importedUsers));
  }

  print('Registering OK.');
}