#145 [feature] We now supports realtime chat with MQTT protocol + new cmd, msg types

[Anonymous]

Originally created by: supreme-gg-gg

Summary

"Two months ago, we set forth on a bold quest to make Instagram CLI great again. Today, we plant our flag upon new ground — Instagram CLI now speaks MQTT, the native tongue of Instagram’s own apps.**

• New commands for media and video upload
• New command to enter selection mode (similar to python UI, see below for difference) + react, unsend
• Hybrid client design (see designs/hybrid-client-design.md), where every request falls back to the traditional API client when MQTT realtime client is not available.
• New event-driven chat interface to work with realtime chat event updates -- we've significantly reduced latency compared to the old polling approach, and much more robust against Meta detection!!
• Refactored type definitions which are then used in the message parser, now shared between the old API and new MQTT because most of the fields are identical. The type definition of the old API is deprecated (e.g. does not contain media) and is now monkey patched 😄, various components are updated as a result of type update
• Added e2e mock test to CI for chat and feed view

Notes

• Workflow: on thread selection we load chat history with old API -> events are streamed and handled by the parser and rendered / updates the chat UI -> any user actions or commands will invoke the realtime client command if available, otherwise use the old API
• Realtime client initialisation is handled by client.ts on login, the view using this client is responsible for calling its destructor manually on unmount / end of life cycle
• I propose to separate selection logic into first select, then act, this means if you wnat to react for example, run :select to select like Python, but then after confirming you run :react emoji again to react to that message. The benefit is that the user can more flexibly choose what to do with that selected message, e.g. if you initially want to reply, but you're too lazy and just wanted to react, here you go 😄
• Also note that you can just run :react without any emoji, it acts like double-tapping (sends like) by default!

No linter errors were added!! Let's gooooo zero technical debt moment?!

Not implemented yet

These will likely not be fixed as part of this PR:

• Receiving reactions via MQTT! Reactions are somehow sent as action logs, not even reactions logs, and this class does not contain any information about which item it belongs to , user ID, emoji, it only contains a text: supreme_gg_gg reacted 😂 to your message. This is impossible for us to make any use of it. We're investigating hoping for a fix someday (not in this PR)
• Reply! Replying to messages is not possible right now because these api clients do not accept arguments to specify reply to ID, see discussion in [#105]

Next steps

• Investigate issues with reactions and reply
• Expand MQTT support there are more hooks that isn't used!
• Add basic Integration tests in CI [#144]
• Make release on NPM
• [#136] might be fixed or might not

[Anonymous]

Originally posted by: supreme-gg-gg

EDIT: Everything in this PR should work now!

@endernoke if you have an idea of reply and receiving reactions lmk otherwise I'd say let it rot I don't have time rn to inspect packages for reaction lol, I tried the most likely ones and they don't seem to show up

Feel free to roast any design + efficiency + vulnerability + technical debt during the review since we want to make sure nothing breaks due to this PR. Take your time I don't plan to merge this soon since I'm busy these few days

[Anonymous]

Originally posted by: supreme-gg-gg

For anyone interested, here is a piece of code showcasing what else is possible to do with MQTT, you can run this as a starting point if you want to understand what we're doing here and if you want to contribute by learning what features it supports:

/* eslint-disable no-console */
import {IgApiClientRealtime, withRealtime} from 'instagram_mqtt';
import {GraphQLSubscriptions} from 'instagram_mqtt/dist/realtime';
import {IgApiClient} from 'instagram-private-api';
import {SkywalkerSubscriptions} from 'instagram_mqtt';
import {existsSync, readFileSync, writeFileSync} from 'fs';
import * as readline from 'node:readline/promises';
import {stdin as input, stdout as output} from 'node:process';

async function getUserInput(prompt: string): Promise<string> {
    const rl = readline.createInterface({input, output});
    const answer = await rl.question(prompt);
    rl.close();
    return answer;
}
(async () => {
    // this extends the IgApiClient with realtime features
    const ig: IgApiClientRealtime = withRealtime(
        new IgApiClient() /* you may pass mixins in here */,
    );
    // regular login
    const IG_USERNAME = await getUserInput('Enter your Instagram username: ');
    const IG_PASSWORD = await getUserInput('Enter your Instagram password: ');
    ig.state.generateDevice(IG_USERNAME);
    await readState(ig);
    ig.request.end$.subscribe(() => saveState(ig));
    console.log(`User: ${IG_USERNAME}`);
    await ig.account.login(IG_USERNAME, IG_PASSWORD);
    await ig.qe.syncLoginExperiments();
    // now `ig` is a client with a valid session

    // whenever something gets sent and has no event, this is called
    ig.realtime.on('receive', (topic, messages) =>
        console.log('receive', topic, messages),
    );

    // this is called with a wrapper use {message} to only get the "actual" message from the wrapper
    ig.realtime.on('message', logEvent('messageWrapper'));

    // a thread is updated, e.g. admins/members added/removed
    ig.realtime.on('threadUpdate', logEvent('threadUpdateWrapper'));

    // other direct messages - no messages
    ig.realtime.on('direct', logEvent('direct'));

    // whenever something gets sent to /ig_realtime_sub and has no event, this is called
    ig.realtime.on('realtimeSub', logEvent('realtimeSub'));

    // whenever the client has a fatal error
    ig.realtime.on('error', console.error);

    ig.realtime.on('close', () => console.error('RealtimeClient closed'));

    // connect
    // this will resolve once all initial subscriptions have been sent
    await ig.realtime.connect({
        // optional
        graphQlSubs: [
            // these are some subscriptions
            GraphQLSubscriptions.getAppPresenceSubscription(),
            GraphQLSubscriptions.getZeroProvisionSubscription(ig.state.phoneId),
            GraphQLSubscriptions.getDirectStatusSubscription(),
            GraphQLSubscriptions.getDirectTypingSubscription(ig.state.cookieUserId),
            GraphQLSubscriptions.getAsyncAdSubscription(ig.state.cookieUserId),
        ],
        // optional
        skywalkerSubs: [
            SkywalkerSubscriptions.directSub(ig.state.cookieUserId),
            SkywalkerSubscriptions.liveSub(ig.state.cookieUserId),
        ],
        // optional
        // this enables you to get direct messages
        irisData: await ig.feed.directInbox().request(),
        // optional
        // in here you can change connect options
        // available are all properties defined in MQTToTConnectionClientInfo
        connectOverrides: {},

        // optional
        // use this proxy
        // socksOptions: {
        //    type: 5,
        //    port: 12345,
        //    host: '...',
        // },
    });

    console.log('Connected to Realtime');

    // Interactive messaging functionality
    await startInteractiveMessaging(ig);

    // simulate turning the device off after 2s and turning it back on after another 2s
    // setTimeout(() => {
    //    console.log('Device off');
    //    // from now on, you won't receive any realtime-data as you "aren't in the app"
    //    // the keepAliveTimeout is somehow a 'constant' by instagram
    //    ig.realtime.direct.sendForegroundState({
    //       inForegroundApp: false,
    //       inForegroundDevice: false,
    //       keepAliveTimeout: 900,
    //    });
    // }, 2000);
    // setTimeout(() => {
    //    console.log('In App');
    //    ig.realtime.direct.sendForegroundState({
    //       inForegroundApp: true,
    //       inForegroundDevice: true,
    //       keepAliveTimeout: 60,
    //    });
    // }, 4000);

    // an example on how to subscribe to live comments
    // you can add other GraphQL subs using .subscribe
    // await ig.realtime.graphQlSubscribe(GraphQLSubscriptions.getLiveRealtimeCommentsSubscription('<broadcast-id>'));
})();

/**

 * Interactive messaging functionality
 * @param ig - The Instagram realtime client
 */
async function startInteractiveMessaging(ig: IgApiClientRealtime) {
    console.log('\n=== Interactive Messaging Started ===');
    console.log('Commands:');
    console.log('  /send <threadId> <message>  - Send a text message');
    console.log('  /like <threadId>            - Send a like/heart');
    console.log('  /typing <threadId>          - Start typing indicator');
    console.log('  /stop-typing <threadId>     - Stop typing indicator');
    console.log('  /seen <threadId> <itemId>   - Mark message as seen');
    console.log('  /react <threadId> <itemId> <emoji> - React with emoji');
    console.log('  /threads                    - List recent threads');
    console.log('  /help                       - Show this help');
    console.log('  /quit                       - Exit\n');

    while (true) {
        try {
            const input = await getUserInput('> ');
            const [command, ...args] = input.trim().split(' ');

            switch (command.toLowerCase()) {
                case '/send':
                    if (args.length < 2) {
                        console.log('Usage: /send <threadId> <message>');
                        break;
                    }
                    const [threadId, ...messageParts] = args;
                    const message = messageParts.join(' ');

                    console.log(`Sending message to thread ${threadId}: "${message}"`);
                    await ig.realtime.direct?.sendText({
                        threadId,
                        text: message,
                    });
                    console.log('✓ Message sent');
                    break;

                case '/like':
                    if (args.length < 1) {
                        console.log('Usage: /like <threadId>');
                        break;
                    }
                    console.log(`Sending like to thread ${args[0]}`);
                    await ig.realtime.direct?.sendLike({
                        threadId: args[0],
                    });
                    console.log('✓ Like sent');
                    break;

                case '/typing':
                    if (args.length < 1) {
                        console.log('Usage: /typing <threadId>');
                        break;
                    }
                    console.log(`Starting typing indicator in thread ${args[0]}`);
                    await ig.realtime.direct?.indicateActivity({
                        threadId: args[0],
                        isActive: true,
                    });
                    console.log('✓ Typing indicator started');
                    break;

                case '/stop-typing':
                    if (args.length < 1) {
                        console.log('Usage: /stop-typing <threadId>');
                        break;
                    }
                    console.log(`Stopping typing indicator in thread ${args[0]}`);
                    await ig.realtime.direct?.indicateActivity({
                        threadId: args[0],
                        isActive: false,
                    });
                    console.log('✓ Typing indicator stopped');
                    break;

                case '/seen':
                    if (args.length < 2) {
                        console.log('Usage: /seen <threadId> <itemId>');
                        break;
                    }
                    console.log(
                        `Marking message ${args[1]} as seen in thread ${args[0]}`,
                    );
                    await ig.realtime.direct?.markAsSeen({
                        threadId: args[0],
                        itemId: args[1],
                    });
                    console.log('✓ Message marked as seen');
                    break;

                case '/react':
                    if (args.length < 3) {
                        console.log('Usage: /react <threadId> <itemId> <emoji>');
                        break;
                    }
                    console.log(
                        `Reacting to message ${args[1]} with ${args[2]} in thread ${args[0]}`,
                    );
                    await ig.realtime.direct?.sendReaction({
                        threadId: args[0],
                        itemId: args[1],
                        emoji: args[2],
                        reactionStatus: 'created',
                    });
                    console.log('✓ Reaction sent');
                    break;

                case '/threads':
                    console.log('Fetching recent threads...');
                    try {
                        const inbox = await ig.feed.directInbox().request();
                        console.log('\nRecent threads:');
                        inbox.inbox?.threads?.slice(0, 10).forEach((thread, index) => {
                            console.log(`  ${index + 1}. Thread ID: ${thread.thread_id}`);
                            console.log(`     Title: ${thread.thread_title || 'No title'}`);
                            console.log(
                                `     Users: ${thread.users?.map(u => u.username).join(', ') || 'Unknown'}`,
                            );
                            const lastActivity = thread.last_activity_at
                                ? new Date(
                                        Number(thread.last_activity_at) / 1000,
                                    ).toLocaleString()
                                : 'Unknown';
                            console.log(`     Last activity: ${lastActivity}\n`);
                        });
                    } catch (error) {
                        console.log('Error fetching threads:', error);
                    }
                    break;

                case '/help':
                    console.log('\nCommands:');
                    console.log('  /send <threadId> <message>  - Send a text message');
                    console.log('  /like <threadId>            - Send a like/heart');
                    console.log('  /typing <threadId>          - Start typing indicator');
                    console.log('  /stop-typing <threadId>     - Stop typing indicator');
                    console.log('  /seen <threadId> <itemId>   - Mark message as seen');
                    console.log(
                        '  /react <threadId> <itemId> <emoji> - React with emoji',
                    );
                    console.log('  /threads                    - List recent threads');
                    console.log('  /help                       - Show this help');
                    console.log('  /quit                       - Exit\n');
                    break;

                case '/quit':
                    console.log('Exiting interactive messaging...');
                    return;

                case '':
                    break;

                default:
                    console.log(
                        `Unknown command: ${command}. Type /help for available commands.`,
                    );
                    break;
            }
        } catch (error) {
            console.error('Error executing command:', error);
        }
    }
}

/**

 * A wrapper function to log to the console
 * @param name
 * @returns {(data) => void}
 */
function logEvent(name: string) {
    return (data: any) => console.log(name, data);
}

async function saveState(ig: IgApiClientRealtime) {
    return writeFileSync('state.json', await ig.exportState(), {
        encoding: 'utf8',
    });
}

async function readState(ig: IgApiClientRealtime) {
    if (!existsSync('state.json')) return;
    await ig.importState(readFileSync('state.json', {encoding: 'utf8'}));
}

Of course, this is far from being comprehensive! You should refer to the original repository listed in [#74]

[Anonymous]

Originally posted by: endernoke

EDIT: Everything in this PR should work now!

@endernoke if you have an idea of reply and receiving reactions lmk otherwise I'd say let it rot I don't have time rn to inspect packages for reaction lol, I tried the most likely ones and they don't seem to show up

Feel free to roast any design + efficiency + vulnerability + technical debt during the review since we want to make sure nothing breaks due to this PR. Take your time I don't plan to merge this soon since I'm busy these few days

@supreme-gg-gg you are the hero and savior of instagram-cli!! With MQTT we have crossed a new milestone in the long march.
I'm being chased by academic debt creditors lol I'll review this in a few days

For anyone interested, here is a piece of code showcasing what else is possible to do with MQTT, you can run this as a starting point if you want to understand what we're doing here and if you want to contribute by learning what features it supports:

```ts
/ eslint-disable no-console /
import {IgApiClientRealtime, withRealtime} from 'instagram_mqtt';
import {GraphQLSubscriptions} from 'instagram_mqtt/dist/realtime';
import {IgApiClient} from 'instagram-private-api';
import {SkywalkerSubscriptions} from 'instagram_mqtt';
import {existsSync, readFileSync, writeFileSync} from 'fs';
import * as readline from 'node:readline/promises';
import {stdin as input, stdout as output} from 'node:process';

async function getUserInput(prompt: string): Promise<string> {
const rl = readline.createInterface({input, output});
const answer = await rl.question(prompt);
rl.close();
return answer;
}
(async () => {
// this extends the IgApiClient with realtime features
const ig: IgApiClientRealtime = withRealtime(
new IgApiClient() / you may pass mixins in here /,
);
// regular login
const IG_USERNAME = await getUserInput('Enter your Instagram username: ');
const IG_PASSWORD = await getUserInput('Enter your Instagram password: ');
ig.state.generateDevice(IG_USERNAME);
await readState(ig);
ig.request.end$.subscribe(() => saveState(ig));
console.log(User: ${IG_USERNAME});
await ig.account.login(IG_USERNAME, IG_PASSWORD);
await ig.qe.syncLoginExperiments();
// now ig is a client with a valid session
// ...
```</string>

Of course, this is far from being comprehensive! You should refer to the original repository listed in [#74]

OMG is this the exact code we vibe coded 2 months ago 😮🤯😨

[Anonymous]

Originally posted by: supreme-gg-gg

indeed, open sourced the vibe coded manuscript now 👀 😄

[Anonymous]

Originally posted by: supreme-gg-gg

LGTM the architecture looks good 👍

I agree with the select -> action flow for interacting with messages. It also makes state management easier.

For message selection, I think the current box border is causing a somewhat large UI change every time the selection moves (the selected message shifts slightly and the border pushes other messages up/down). Perhaps consider using bold text or inversed color to indicate selection. Feel free to keep the current UI if you think it's fine.

I think it's nice to have some "animation" effect when the user moves the selection box, this is "trying" to do it, though I'm no UX engineer so you might be right lol 😭 I will find out a better way with the next UI patch PR before release

[Anonymous]

Ticket changed by: supreme-gg-gg

• status: open --> closed