save message sync
This commit is contained in:
grimhilt
parent
f9fbab3a21
commit
95f39cf53a
@ -35,6 +35,7 @@ if (shouldReset) {
|
||||
return;
|
||||
}
|
||||
|
||||
|
||||
imap.once("ready", function () {
|
||||
const readOnly = true;
|
||||
imap.openBox("INBOX", readOnly, (err, box) => {
|
||||
@ -45,32 +46,18 @@ imap.once("ready", function () {
|
||||
|
||||
const f = imap.fetch(970, {
|
||||
size: true,
|
||||
struct: true,
|
||||
envelope: true,
|
||||
});
|
||||
let messageIDs = [];
|
||||
// var f = imap.seq.fetch('1:3', {
|
||||
// bodies: 'HEADER.FIELDS (FROM TO SUBJECT DATE)',
|
||||
// struct: true
|
||||
// });
|
||||
f.on("message", function (msg, seqno) {
|
||||
// console.log("Message #%d", seqno);
|
||||
// var prefix = "(#" + seqno + ") ";
|
||||
// msg.on("body", function (stream, info) {
|
||||
// simpleParser(stream, async (err, parsed) => {
|
||||
// // find box id;
|
||||
// // console.log(parsed)
|
||||
// const boxId = 1;
|
||||
// saveMessage(parsed, boxId);
|
||||
// // console.log(parsed.subject);
|
||||
// // fs.writeFileSync("./test.txt", JSON.stringify(parsed));
|
||||
// });
|
||||
// // console.log(prefix + 'Body');
|
||||
// // stream.pipe(fs.createWriteStream('msg-' + seqno + '-body.txt'));
|
||||
// });
|
||||
msg.once("attributes", (attrs) => {
|
||||
// todo find boxId
|
||||
const boxId = 1;
|
||||
// console.log(attrs)
|
||||
messageIDs.push(attrs.envelope.messageId)
|
||||
saveMessage(attrs, boxId, imap);
|
||||
});
|
||||
});
|
||||
@ -79,71 +66,11 @@ imap.once("ready", function () {
|
||||
console.log("Fetch error: " + err);
|
||||
});
|
||||
f.once("end", function () {
|
||||
console.log(messageIDs)
|
||||
isDone = true;
|
||||
console.log("Done fetching all messages!");
|
||||
// imap.end();
|
||||
});
|
||||
// });
|
||||
return;
|
||||
// if (err) throw err;
|
||||
// const f = imap.seq.fetch('2:2', {
|
||||
// bodies: ['HEADER.FIELDS (FROM)','TEXT'],
|
||||
// struct: true,
|
||||
// envelope: true,
|
||||
// extensions: true
|
||||
// });
|
||||
|
||||
// f.on('message', function(msg, seqno) {
|
||||
// // console.log('Message #%d', seqno);
|
||||
// var prefix = '(#' + seqno + ') ';
|
||||
// let attributes = undefined;
|
||||
// let body = undefined;
|
||||
|
||||
// msg.on('body', function(stream, info) {
|
||||
// simpleParser(stream, async (err, parsed) => {
|
||||
// body = parsed;
|
||||
// // console.log(body)
|
||||
// if (attributes) {
|
||||
// saveMessage(body, attributes);
|
||||
// };
|
||||
// // console.log(parsed.headers)
|
||||
// // const {from, subject, textAsHtml, text} = parsed;
|
||||
// // console.log(parsed.attachments)
|
||||
// // console.log(prefix + parsed.text)
|
||||
// // console.log(parsed.from.value);
|
||||
// // console.log(parsed.subject);
|
||||
// // console.log(parsed.date)
|
||||
// // console.log(parsed.replyTo.value);
|
||||
// // console.log(parsed.messageId);
|
||||
// // console.log(parsed.html);
|
||||
// // console.log(parsed.text);
|
||||
// // console.log(parsed.textAsHtml);
|
||||
// });
|
||||
// });
|
||||
|
||||
// msg.once('attributes', attrs => {
|
||||
// attributes = attrs;
|
||||
// console.log(attributes)
|
||||
// if (body) {
|
||||
// saveMessage(body, attributes);
|
||||
// };
|
||||
// // console.log(prefix + 'Attributes: %s', inspect(attrs, false, 8));
|
||||
|
||||
// });
|
||||
|
||||
// msg.once('end', function() {
|
||||
// console.log(prefix + 'Finished');
|
||||
// });
|
||||
|
||||
// });
|
||||
|
||||
// f.once('error', function(err) {
|
||||
// console.log('Fetch error: ' + err);
|
||||
// });
|
||||
|
||||
// f.once('end', function() {
|
||||
// console.log('Done fetching all messages!');
|
||||
// imap.end();
|
||||
// });
|
||||
// imap.end()
|
||||
});
|
||||
});
|
||||
|
||||
|
||||
@ -2,6 +2,7 @@ const { getAddresseId } = require("../sql/mail");
|
||||
const { DEBUG } = require("../utils/debug");
|
||||
const { simpleParser } = require("mailparser");
|
||||
const moment = require("moment");
|
||||
const fs = require("fs");
|
||||
const {
|
||||
registerMessage,
|
||||
registerMailbox_message,
|
||||
@ -9,6 +10,7 @@ const {
|
||||
saveAddress_fields,
|
||||
registerBodypart,
|
||||
saveBodypart,
|
||||
saveSource,
|
||||
} = require("../sql/saveMessage");
|
||||
|
||||
const {
|
||||
@ -24,13 +26,14 @@ const { getFieldId, findRoomByOwner } = require("../sql/mail");
|
||||
|
||||
function saveMessage(attrs, mailboxId, imap) {
|
||||
const envelope = attrs.envelope;
|
||||
const timestamp = moment(new Date(envelope.date).getTime()).format(
|
||||
const ts = moment(new Date(envelope.date).getTime()).format(
|
||||
"YYYY-MM-DD HH:mm:ss"
|
||||
);
|
||||
const rfc822size = attrs.size;
|
||||
const messageID = envelope.messageId;
|
||||
|
||||
registerMessage(timestamp, rfc822size, messageID).then((messageId) => {
|
||||
registerMessage(ts, rfc822size, messageID)
|
||||
.then((messageId) => {
|
||||
const isSeen = attrs.flags.includes("Seen") ? 1 : 0; // todo verify
|
||||
const deleted = attrs.flags.includes("Deleted") ? 1 : 0; // todo verify
|
||||
|
||||
@ -42,96 +45,89 @@ function saveMessage(attrs, mailboxId, imap) {
|
||||
isSeen,
|
||||
deleted
|
||||
);
|
||||
const f = imap.fetch(attrs.uid, { bodies: "" });
|
||||
let buffer = "";
|
||||
|
||||
const len = attrs.struct.length;
|
||||
attrs.struct.forEach((part) => {
|
||||
|
||||
if (len > 1) part = part[0];
|
||||
// todo should be recursive to take eveyraçpghyrue
|
||||
// todo attachments
|
||||
console.log("parr", part)
|
||||
if (part?.type == "text") {
|
||||
console.log(attrs.uid, part.partID)
|
||||
const fetch = imap.fetch(attrs.uid, { bodies: ['', part.partID] });
|
||||
|
||||
fetch.on("message", (msg) => {
|
||||
msg.on("body", (stream, info) => {
|
||||
simpleParser(stream, async (err, parsed) => {
|
||||
console.log(part.partID, parsed?.subject);
|
||||
});
|
||||
f.on("message", function (msg, seqno) {
|
||||
msg.on("body", function (stream, info) {
|
||||
stream.on("data", function (chunk) {
|
||||
buffer += chunk.toString("utf8");
|
||||
});
|
||||
|
||||
msg.once("end", () => {
|
||||
console.log("Finished fetching message");
|
||||
});
|
||||
stream.once("end", () => {
|
||||
// save raw data
|
||||
saveSource(messageId, buffer);
|
||||
|
||||
// parse data
|
||||
simpleParser(buffer, async (err, parsed) => {
|
||||
saveFromParsedData(parsed, messageId);
|
||||
});
|
||||
fetch.once("error", (err) => {
|
||||
console.log(err);
|
||||
});
|
||||
});
|
||||
});
|
||||
f.once("error", function (err) {
|
||||
console.log("Fetch error: " + err);
|
||||
});
|
||||
f.once("end", function () {
|
||||
console.log("Done fetching all messages!");
|
||||
});
|
||||
})
|
||||
.catch((err) => {
|
||||
DEBUG.log("Unable to register message: " + err);
|
||||
});
|
||||
}
|
||||
|
||||
fetch.once("end", () => {
|
||||
console.log("Done fetching all messages");
|
||||
function saveFromParsedData(parsed, messageId) {
|
||||
fs.writeFileSync("./test.txt", JSON.stringify(parsed));
|
||||
Object.keys(parsed).forEach((key) => {
|
||||
if (["from", "to", "cc", "bcc", "reply-to"].includes(key)) {
|
||||
// save address field
|
||||
getFieldId(key).then((fieldId) => {
|
||||
parsed[key].value.forEach((addr, nb) => {
|
||||
getAddresseId(addr.address, addr.name).then((addressId) => {
|
||||
saveAddress_fields(messageId, fieldId, addressId, nb);
|
||||
});
|
||||
const registerType = `${part.type}/${part.subtype}`;
|
||||
const hash = "2"; // todo
|
||||
const text = "1"; // todo
|
||||
saveBodypart(part.size, hash, text, "").then((bodypartId) => {
|
||||
registerBodypart(
|
||||
});
|
||||
});
|
||||
} else if (["subject", "inReplyTo"].includes(key)) {
|
||||
// todo : "references"
|
||||
getFieldId(key).then((fieldId) => {
|
||||
saveHeader_fields(
|
||||
messageId,
|
||||
registerType,
|
||||
bodypartId,
|
||||
part.size,
|
||||
part.lines
|
||||
fieldId,
|
||||
undefined,
|
||||
undefined,
|
||||
parsed[key]
|
||||
);
|
||||
});
|
||||
}
|
||||
});
|
||||
|
||||
|
||||
|
||||
// todo when transfered
|
||||
//The part column records to which MIME part this header field belongs. It's empty for the main header (the one seen above) and nonempty when a multipart message has headers on each part.
|
||||
const part = ""; // todo ^
|
||||
// save envelope (header + from, to, subject, date, cc)
|
||||
// Object.keys(envelope).forEach((key, position) => {
|
||||
// const newKey = keyNormalizer(key);
|
||||
// if (isHeader(newKey)) {
|
||||
// getFieldId(newKey).then((fieldId) => {
|
||||
} else if (["html", "text", "textAsHtml"].includes(key)) {
|
||||
const hash = "0";
|
||||
const size = "0";
|
||||
// saveBodypart(size, hash, parsed[key], "").then((bodypartId) => {
|
||||
// getFieldId(key).then((fieldId) => {
|
||||
// saveHeader_fields(
|
||||
// messageId,
|
||||
// part,
|
||||
// position,
|
||||
// fieldId,
|
||||
// envelope[key]
|
||||
// );
|
||||
// });
|
||||
// } else {
|
||||
// getFieldId(newKey).then((fieldId) => {
|
||||
// if (!envelope[key]) {
|
||||
// return;
|
||||
// }
|
||||
// envelope[key].forEach((elt, index) => {
|
||||
// getAddresseId(createAddress(elt)).then((addressId) => {
|
||||
// saveAddress_fields(
|
||||
// messageId,
|
||||
// part,
|
||||
// position,
|
||||
// fieldId,
|
||||
// index,
|
||||
// addressId
|
||||
// bodypartId,
|
||||
// undefined, // todo ?
|
||||
// undefined
|
||||
// );
|
||||
// });
|
||||
// });
|
||||
// });
|
||||
// }
|
||||
// });
|
||||
|
||||
// todo check for different provider name of inreplyto
|
||||
// registerMessageInApp(envelope, messageId, isSeen);
|
||||
}).catch((err) => {
|
||||
DEBUG.log("Unable to register message: "+err);
|
||||
} else if (key == "attachments") {
|
||||
// todo
|
||||
} else if (
|
||||
["date", "messageId", "headers", "headerLines"].includes(key)
|
||||
) {
|
||||
// messageId and date are already saved
|
||||
// other field are not improted and can be retrieved in source
|
||||
return;
|
||||
} else {
|
||||
DEBUG.log("doesn't know key: " + key);
|
||||
return;
|
||||
}
|
||||
});
|
||||
// todo when transfered
|
||||
}
|
||||
|
||||
function haveSameReceivers() {
|
||||
@ -152,7 +148,6 @@ function registerMessageInApp(envelope, messageId, isSeen) {
|
||||
} else {
|
||||
findRoomByOwner(ownerId).then((res) => {
|
||||
if (res.length == 0) {
|
||||
console.log(res, ownerId)
|
||||
createRoom(envelope.subject, ownerId).then((roomId) => {
|
||||
registerMessageInRoom(messageId, roomId, isSeen);
|
||||
});
|
||||
@ -225,38 +220,6 @@ function registerReplyMessage(envelope, messageId, isSeen) {
|
||||
// `SELECT app_room_messages.room, app_room_messages.thread FROM app_room_messages INNER JOIN messages WHERE messages.messageID = '${envelope.inReplyTo}' AND app_room_messages.message = messages.id`;
|
||||
}
|
||||
|
||||
function findTextPart(struct) {
|
||||
for (var i = 0, len = struct.length, r; i < len; ++i) {
|
||||
if (Array.isArray(struct[i])) {
|
||||
if ((r = findTextPart(struct[i]))) return r;
|
||||
} else if (
|
||||
struct[i].type === "text" &&
|
||||
(struct[i].subtype === "plain" || struct[i].subtype === "html")
|
||||
)
|
||||
return [struct[i].partID, struct[i].type + "/" + struct[i].subtype];
|
||||
}
|
||||
}
|
||||
|
||||
function isHeader(key) {
|
||||
switch (key) {
|
||||
case "date":
|
||||
case "subject":
|
||||
case "messageId":
|
||||
case "inReplyTo": // when transfer or reply to message
|
||||
return true;
|
||||
case "from":
|
||||
case "sender":
|
||||
case "replyTo":
|
||||
case "to":
|
||||
case "cc":
|
||||
case "bcc":
|
||||
return false;
|
||||
default:
|
||||
DEBUG.log("Unknown header key: " + key);
|
||||
return true;
|
||||
}
|
||||
}
|
||||
|
||||
function keyNormalizer(key) {
|
||||
// todo
|
||||
return key;
|
||||
|
||||
@ -10,10 +10,9 @@ async function registerMessage(timestamp, rfc822size, messageId) {
|
||||
}
|
||||
|
||||
function registerMailbox_message(mailboxId, uid, messageId, modseq, seen, deleted) {
|
||||
const query = `INSERT IGNORE INTO mailbox_message (mailbox_id, uid, message_id, modseq, seen, deleted) VALUES ('${mailboxId}', '${uid}', '${messageId}', '${modseq}', '${seen}', '${deleted}')`;
|
||||
bdd.query(query, (err, results, fields) => {
|
||||
if (err) DEBUG.log(err);
|
||||
});
|
||||
const query = `INSERT IGNORE INTO mailbox_message (mailbox_id, uid, message_id, modseq, seen, deleted) VALUES (?, ?, ?, ?, ?, ?)`;
|
||||
const values = [mailboxId, uid, messageId, modseq, seen, deleted];
|
||||
execQuery(query, values);
|
||||
}
|
||||
|
||||
function registerBodypart(messageId, part, bodypartId, bytes, nbLines) {
|
||||
@ -23,27 +22,26 @@ function registerBodypart(messageId, part, bodypartId, bytes, nbLines) {
|
||||
});
|
||||
}
|
||||
|
||||
|
||||
function saveBodypart(bytes, hash, text, data) {
|
||||
return new Promise((resolve, reject) => {
|
||||
const query = `INSERT IGNORE INTO bodypart (bytes, hash, text, data) VALUES ('${bytes}', '${hash}', '${text}', '${data}')`;
|
||||
bdd.query(query, (err, results, fields) => {
|
||||
if (err) reject(err);
|
||||
resolve(results.insertId);
|
||||
});
|
||||
});
|
||||
async function saveBodypart(bytes, hash, text, data) {
|
||||
const query = `INSERT IGNORE INTO bodypart (bytes, hash, text, data) VALUES (?, ?, ?,)`;
|
||||
const values = [bytes, hash, text, data];
|
||||
return await execQueryAsyncWithId(query, values);
|
||||
}
|
||||
|
||||
function saveHeader_fields(message, part, position, field, value) {
|
||||
const query = `INSERT INTO header_field (message_id, part, position, field_id, value) VALUES (?, ?, ?, ?, ?)`;
|
||||
const values = [message, part, position, field, value];
|
||||
execQuery(query, values);
|
||||
async function saveHeader_fields(messageId, fieldId, bodypartId, part, value) {
|
||||
const query = `INSERT IGNORE INTO header_field (message_id, field_id, bodypart_id, part, value) VALUES (?, ?, ?, ?, ?)`;
|
||||
const values = [messageId, fieldId, bodypartId, part, value];
|
||||
return await execQueryAsync(query, values);
|
||||
}
|
||||
|
||||
function saveAddress_fields(message, part, position, field, number, address) {
|
||||
const query = `INSERT INTO address_field (message_id , part, position, field_id, number, address_id) VALUES (?, ?, ?, ?, ?, ?)`;
|
||||
const values = [message, part, position, field, number, address];
|
||||
execQuery(query, values);
|
||||
async function saveAddress_fields(messageId, fieldId, addressId, number) {
|
||||
const query = `INSERT IGNORE INTO address_field (message_id , field_id, address_id, number) VALUES (?, ?, ?, ?)`;
|
||||
const values = [messageId, fieldId, addressId, number];
|
||||
return execQueryAsync(query, values);
|
||||
}
|
||||
|
||||
function saveSource(messageId, content) {
|
||||
// todo
|
||||
}
|
||||
|
||||
module.exports = {
|
||||
@ -53,4 +51,5 @@ module.exports = {
|
||||
saveAddress_fields,
|
||||
registerBodypart,
|
||||
saveBodypart,
|
||||
saveSource
|
||||
}
|
||||
@ -139,3 +139,19 @@ create table app_accounts (
|
||||
tls int(1) not null default 0,
|
||||
primary key (id)
|
||||
);
|
||||
|
||||
create table app_rooms (
|
||||
id int not null auto_increment,
|
||||
name text not null,
|
||||
owner int not null,
|
||||
isGroup BIT(1) not null default 0,
|
||||
notSeen int not null default 0,
|
||||
lastUpdate timestamp not null,
|
||||
primary key (id)
|
||||
);
|
||||
|
||||
create table app_room_messages (
|
||||
message int [not null]
|
||||
room int,
|
||||
primary key()
|
||||
)
|
||||
@ -74,15 +74,11 @@ CREATE TABLE bodypart (
|
||||
);
|
||||
|
||||
-- 7
|
||||
CREATE TABLE part_number (
|
||||
CREATE TABLE source (
|
||||
message_id INT NOT NULL,
|
||||
part VARCHAR(128) NOT NULL,
|
||||
bodypart_id INT NOT NULL,
|
||||
bytes INT,
|
||||
nb_lines INT,
|
||||
PRIMARY KEY (message_id, part),
|
||||
FOREIGN KEY (message_id) REFERENCES message(message_id) ON DELETE CASCADE,
|
||||
FOREIGN KEY (bodypart_id) REFERENCES bodypart(bodypart_id) ON DELETE CASCADE
|
||||
content TEXT NOT NULL,
|
||||
PRIMARY KEY (message_id),
|
||||
FOREIGN KEY (message_id) REFERENCES message(message_id) ON DELETE CASCADE
|
||||
);
|
||||
|
||||
-- 8
|
||||
@ -96,24 +92,24 @@ CREATE TABLE field_name (
|
||||
-- 9
|
||||
CREATE TABLE header_field (
|
||||
message_id INT NOT NULL,
|
||||
part VARCHAR(128) NOT NULL,
|
||||
position INT NOT NULL,
|
||||
field_id INT NOT NULL,
|
||||
bodypart_id INT,
|
||||
part VARCHAR(128),
|
||||
value TEXT NOT NULL,
|
||||
UNIQUE KEY (message_id, part, position, field_id),
|
||||
FOREIGN KEY (message_id, part) REFERENCES part_number(message_id, part) ON DELETE CASCADE,
|
||||
FOREIGN KEY (field_id) REFERENCES field_name(field_id)
|
||||
UNIQUE KEY (message_id, field_id, bodypart_id),
|
||||
FOREIGN KEY (message_id) REFERENCES message(message_id) ON DELETE CASCADE,
|
||||
FOREIGN KEY (field_id) REFERENCES field_name(field_id), -- todo on delete behavior
|
||||
FOREIGN KEY (bodypart_id) REFERENCES bodypart(bodypart_id)
|
||||
);
|
||||
|
||||
-- 10
|
||||
CREATE TABLE address_field (
|
||||
message_id INT NOT NULL,
|
||||
part VARCHAR(128) NOT NULL,
|
||||
position INT NOT NULL,
|
||||
field_id INT NOT NULL,
|
||||
number INT,
|
||||
address_id INT NOT NULL,
|
||||
FOREIGN KEY (message_id, part) REFERENCES part_number(message_id, part) ON DELETE CASCADE,
|
||||
number INT,
|
||||
UNIQUE KEY (message_id, field_id, address_id),
|
||||
FOREIGN KEY (message_id) REFERENCES message(message_id) ON DELETE CASCADE,
|
||||
FOREIGN KEY (field_id) REFERENCES field_name(field_id),
|
||||
FOREIGN KEY (address_id) REFERENCES address(address_id)
|
||||
);
|
||||
|
||||
Loading…
Reference in New Issue
Block a user