temp.js

require('dotenv').config();

const {WebClient} = require('@slack/client');
const AWS = require('aws-sdk');
const moment = require('moment');
const {groupBy} = require('lodash');

(async () => {
	const slack = new WebClient(process.env.SLACK_TOKEN);

	const db = new AWS.DynamoDB({
		region: 'ap-northeast-1',
	});
	const doc = new AWS.DynamoDB.DocumentClient({
		region: 'ap-northeast-1',
		convertEmptyValues: true,
	});
	const {TableNames: tables} = await db.listTables({}).promise();

	try {
		let latest;
		while (true) {
			const data = await slack.channels.history({
				channel: 'C7AAX50QY',
				count: 1000,
				latest,
			});

			console.log(`Retrieved ${data.messages.length} rows (latest = ${latest})`);

			if (data.messages.length === 0) {
				break;
			}

			const messages = data.messages.filter((message) => message.type === 'message');

			let offset = 0;
			while (messages.length > offset) {
				const writingMessages = messages.slice(offset, offset + 25);
				console.log(`Inserting ${writingMessages.length} rows... (offset = ${offset})`);

				let interval = 2000;

				while (true) {
					const requestItems = groupBy(writingMessages.map((message) => ({
						PutRequest: {
							Item: message,
						},
					})), (d) => {
						const time = moment(parseFloat(d.PutRequest.Item.ts) * 1000).utcOffset(9);
						return `sandbox-messages-${time.format('YYYYMM')}`;
					});

					const requestTables = Object.keys(requestItems);
					for (const table of requestTables) {
						if (!tables.includes(table)) {
							console.log(`Creating table ${table}...`);

							await db.createTable({
								AttributeDefinitions: [
									{
										AttributeName: 'ts',
										AttributeType: 'S',
									},
								],
								KeySchema: [
									{
										AttributeName: 'ts',
										KeyType: 'HASH',
									},
								],
								TableName: table,
								BillingMode: 'PAY_PER_REQUEST',
							}).promise();
							tables.push(table);

							await new Promise((resolve) => setTimeout(resolve, 1000));
						}
					}

					const response = await doc.batchWrite({
						RequestItems: requestItems,
					}).promise().catch((e) => {
						console.log(e);
						return {
						UnprocessedItems: {hoge: 1},
					}});

					if (Object.keys(response.UnprocessedItems || {}).length === 0) {
						break;
					}

					console.log(`Retrying write operation after ${interval}ms...`);

					// eslint-disable-next-line no-loop-func
					await new Promise((resolve) => setTimeout(resolve, interval));

					interval *= 2;
				}

				offset += writingMessages.length;

				await new Promise((resolve) => setTimeout(resolve, 100));
			}

			const now = Date.now();
			const oneDayBefore = now - 1 * 24 * 60 * 60 * 1000;
			for (const message of messages) {
				const ts = parseFloat(message.ts) * 1000;
				if (ts < oneDayBefore) {
					console.log(`Deleting ${message.ts}`);
					await slack.chat.delete({
						ts: message.ts,
						channel: 'C7AAX50QY',
					});
				}
			}

			if (!data.has_more) {
				break;
			}

			latest = data.messages[data.messages.length - 1].ts;

			await new Promise((resolve) => setTimeout(resolve, 1000));
		}
	} catch (e) {
		console.error(e);
	}
})();