@ -0,0 +1,55 @@ | |||
# frozen_string_literal: true | |||
class Api::V1::ConversationsController < Api::BaseController | |||
LIMIT = 20 | |||
before_action -> { doorkeeper_authorize! :read, :'read:statuses' } | |||
before_action :require_user! | |||
after_action :insert_pagination_headers | |||
respond_to :json | |||
def index | |||
@conversations = paginated_conversations | |||
render json: @conversations, each_serializer: REST::ConversationSerializer | |||
end | |||
private | |||
def paginated_conversations | |||
AccountConversation.where(account: current_account) | |||
.paginate_by_id(limit_param(LIMIT), params_slice(:max_id, :since_id, :min_id)) | |||
end | |||
def insert_pagination_headers | |||
set_pagination_headers(next_path, prev_path) | |||
end | |||
def next_path | |||
if records_continue? | |||
api_v1_conversations_url pagination_params(max_id: pagination_max_id) | |||
end | |||
end | |||
def prev_path | |||
unless @conversations.empty? | |||
api_v1_conversations_url pagination_params(min_id: pagination_since_id) | |||
end | |||
end | |||
def pagination_max_id | |||
@conversations.last.last_status_id | |||
end | |||
def pagination_since_id | |||
@conversations.first.last_status_id | |||
end | |||
def records_continue? | |||
@conversations.size == limit_param(LIMIT) | |||
end | |||
def pagination_params(core_params) | |||
params.slice(:limit).permit(:limit).merge(core_params) | |||
end | |||
end |
@ -0,0 +1,59 @@ | |||
import api, { getLinks } from '../api'; | |||
import { | |||
importFetchedAccounts, | |||
importFetchedStatuses, | |||
importFetchedStatus, | |||
} from './importer'; | |||
export const CONVERSATIONS_FETCH_REQUEST = 'CONVERSATIONS_FETCH_REQUEST'; | |||
export const CONVERSATIONS_FETCH_SUCCESS = 'CONVERSATIONS_FETCH_SUCCESS'; | |||
export const CONVERSATIONS_FETCH_FAIL = 'CONVERSATIONS_FETCH_FAIL'; | |||
export const CONVERSATIONS_UPDATE = 'CONVERSATIONS_UPDATE'; | |||
export const expandConversations = ({ maxId } = {}) => (dispatch, getState) => { | |||
dispatch(expandConversationsRequest()); | |||
const params = { max_id: maxId }; | |||
if (!maxId) { | |||
params.since_id = getState().getIn(['conversations', 0, 'last_status']); | |||
} | |||
api(getState).get('/api/v1/conversations', { params }) | |||
.then(response => { | |||
const next = getLinks(response).refs.find(link => link.rel === 'next'); | |||
dispatch(importFetchedAccounts(response.data.reduce((aggr, item) => aggr.concat(item.accounts), []))); | |||
dispatch(importFetchedStatuses(response.data.map(item => item.last_status).filter(x => !!x))); | |||
dispatch(expandConversationsSuccess(response.data, next ? next.uri : null)); | |||
}) | |||
.catch(err => dispatch(expandConversationsFail(err))); | |||
}; | |||
export const expandConversationsRequest = () => ({ | |||
type: CONVERSATIONS_FETCH_REQUEST, | |||
}); | |||
export const expandConversationsSuccess = (conversations, next) => ({ | |||
type: CONVERSATIONS_FETCH_SUCCESS, | |||
conversations, | |||
next, | |||
}); | |||
export const expandConversationsFail = error => ({ | |||
type: CONVERSATIONS_FETCH_FAIL, | |||
error, | |||
}); | |||
export const updateConversations = conversation => dispatch => { | |||
dispatch(importFetchedAccounts(conversation.accounts)); | |||
if (conversation.last_status) { | |||
dispatch(importFetchedStatus(conversation.last_status)); | |||
} | |||
dispatch({ | |||
type: CONVERSATIONS_UPDATE, | |||
conversation, | |||
}); | |||
}; |
@ -0,0 +1,85 @@ | |||
import React from 'react'; | |||
import PropTypes from 'prop-types'; | |||
import ImmutablePropTypes from 'react-immutable-proptypes'; | |||
import ImmutablePureComponent from 'react-immutable-pure-component'; | |||
import StatusContent from '../../../components/status_content'; | |||
import RelativeTimestamp from '../../../components/relative_timestamp'; | |||
import DisplayName from '../../../components/display_name'; | |||
import Avatar from '../../../components/avatar'; | |||
import AttachmentList from '../../../components/attachment_list'; | |||
import { HotKeys } from 'react-hotkeys'; | |||
export default class Conversation extends ImmutablePureComponent { | |||
static contextTypes = { | |||
router: PropTypes.object, | |||
}; | |||
static propTypes = { | |||
conversationId: PropTypes.string.isRequired, | |||
accounts: ImmutablePropTypes.list.isRequired, | |||
lastStatus: ImmutablePropTypes.map.isRequired, | |||
onMoveUp: PropTypes.func, | |||
onMoveDown: PropTypes.func, | |||
}; | |||
handleClick = () => { | |||
if (!this.context.router) { | |||
return; | |||
} | |||
const { lastStatus } = this.props; | |||
this.context.router.history.push(`/statuses/${lastStatus.get('id')}`); | |||
} | |||
handleHotkeyMoveUp = () => { | |||
this.props.onMoveUp(this.props.conversationId); | |||
} | |||
handleHotkeyMoveDown = () => { | |||
this.props.onMoveDown(this.props.conversationId); | |||
} | |||
render () { | |||
const { accounts, lastStatus, lastAccount } = this.props; | |||
if (lastStatus === null) { | |||
return null; | |||
} | |||
const handlers = { | |||
moveDown: this.handleHotkeyMoveDown, | |||
moveUp: this.handleHotkeyMoveUp, | |||
open: this.handleClick, | |||
}; | |||
let media; | |||
if (lastStatus.get('media_attachments').size > 0) { | |||
media = <AttachmentList compact media={lastStatus.get('media_attachments')} />; | |||
} | |||
return ( | |||
<HotKeys handlers={handlers}> | |||
<div className='conversation focusable' tabIndex='0' onClick={this.handleClick} role='button'> | |||
<div className='conversation__header'> | |||
<div className='conversation__avatars'> | |||
<div>{accounts.map(account => <Avatar key={account.get('id')} size={36} account={account} />)}</div> | |||
</div> | |||
<div className='conversation__time'> | |||
<RelativeTimestamp timestamp={lastStatus.get('created_at')} /> | |||
<br /> | |||
<DisplayName account={lastAccount} withAcct={false} /> | |||
</div> | |||
</div> | |||
<StatusContent status={lastStatus} onClick={this.handleClick} /> | |||
{media} | |||
</div> | |||
</HotKeys> | |||
); | |||
} | |||
} |
@ -0,0 +1,68 @@ | |||
import React from 'react'; | |||
import PropTypes from 'prop-types'; | |||
import ImmutablePropTypes from 'react-immutable-proptypes'; | |||
import ImmutablePureComponent from 'react-immutable-pure-component'; | |||
import ConversationContainer from '../containers/conversation_container'; | |||
import ScrollableList from '../../../components/scrollable_list'; | |||
import { debounce } from 'lodash'; | |||
export default class ConversationsList extends ImmutablePureComponent { | |||
static propTypes = { | |||
conversationIds: ImmutablePropTypes.list.isRequired, | |||
hasMore: PropTypes.bool, | |||
isLoading: PropTypes.bool, | |||
onLoadMore: PropTypes.func, | |||
shouldUpdateScroll: PropTypes.func, | |||
}; | |||
getCurrentIndex = id => this.props.conversationIds.indexOf(id) | |||
handleMoveUp = id => { | |||
const elementIndex = this.getCurrentIndex(id) - 1; | |||
this._selectChild(elementIndex); | |||
} | |||
handleMoveDown = id => { | |||
const elementIndex = this.getCurrentIndex(id) + 1; | |||
this._selectChild(elementIndex); | |||
} | |||
_selectChild (index) { | |||
const element = this.node.node.querySelector(`article:nth-of-type(${index + 1}) .focusable`); | |||
if (element) { | |||
element.focus(); | |||
} | |||
} | |||
setRef = c => { | |||
this.node = c; | |||
} | |||
handleLoadOlder = debounce(() => { | |||
const last = this.props.conversationIds.last(); | |||
if (last) { | |||
this.props.onLoadMore(last); | |||
} | |||
}, 300, { leading: true }) | |||
render () { | |||
const { conversationIds, onLoadMore, ...other } = this.props; | |||
return ( | |||
<ScrollableList {...other} onLoadMore={onLoadMore && this.handleLoadOlder} scrollKey='direct' ref={this.setRef}> | |||
{conversationIds.map(item => ( | |||
<ConversationContainer | |||
key={item} | |||
conversationId={item} | |||
onMoveUp={this.handleMoveUp} | |||
onMoveDown={this.handleMoveDown} | |||
/> | |||
))} | |||
</ScrollableList> | |||
); | |||
} | |||
} |
@ -0,0 +1,15 @@ | |||
import { connect } from 'react-redux'; | |||
import Conversation from '../components/conversation'; | |||
const mapStateToProps = (state, { conversationId }) => { | |||
const conversation = state.getIn(['conversations', 'items']).find(x => x.get('id') === conversationId); | |||
const lastStatus = state.getIn(['statuses', conversation.get('last_status')], null); | |||
return { | |||
accounts: conversation.get('accounts').map(accountId => state.getIn(['accounts', accountId], null)), | |||
lastStatus, | |||
lastAccount: lastStatus === null ? null : state.getIn(['accounts', lastStatus.get('account')], null), | |||
}; | |||
}; | |||
export default connect(mapStateToProps)(Conversation); |
@ -0,0 +1,15 @@ | |||
import { connect } from 'react-redux'; | |||
import ConversationsList from '../components/conversations_list'; | |||
import { expandConversations } from '../../../actions/conversations'; | |||
const mapStateToProps = state => ({ | |||
conversationIds: state.getIn(['conversations', 'items']).map(x => x.get('id')), | |||
isLoading: state.getIn(['conversations', 'isLoading'], true), | |||
hasMore: state.getIn(['conversations', 'hasMore'], false), | |||
}); | |||
const mapDispatchToProps = dispatch => ({ | |||
onLoadMore: maxId => dispatch(expandConversations({ maxId })), | |||
}); | |||
export default connect(mapStateToProps, mapDispatchToProps)(ConversationsList); |
@ -0,0 +1,79 @@ | |||
import { Map as ImmutableMap, List as ImmutableList } from 'immutable'; | |||
import { | |||
CONVERSATIONS_FETCH_REQUEST, | |||
CONVERSATIONS_FETCH_SUCCESS, | |||
CONVERSATIONS_FETCH_FAIL, | |||
CONVERSATIONS_UPDATE, | |||
} from '../actions/conversations'; | |||
import compareId from '../compare_id'; | |||
const initialState = ImmutableMap({ | |||
items: ImmutableList(), | |||
isLoading: false, | |||
hasMore: true, | |||
}); | |||
const conversationToMap = item => ImmutableMap({ | |||
id: item.id, | |||
accounts: ImmutableList(item.accounts.map(a => a.id)), | |||
last_status: item.last_status.id, | |||
}); | |||
const updateConversation = (state, item) => state.update('items', list => { | |||
const index = list.findIndex(x => x.get('id') === item.id); | |||
const newItem = conversationToMap(item); | |||
if (index === -1) { | |||
return list.unshift(newItem); | |||
} else { | |||
return list.set(index, newItem); | |||
} | |||
}); | |||
const expandNormalizedConversations = (state, conversations, next) => { | |||
let items = ImmutableList(conversations.map(conversationToMap)); | |||
return state.withMutations(mutable => { | |||
if (!items.isEmpty()) { | |||
mutable.update('items', list => { | |||
list = list.map(oldItem => { | |||
const newItemIndex = items.findIndex(x => x.get('id') === oldItem.get('id')); | |||
if (newItemIndex === -1) { | |||
return oldItem; | |||
} | |||
const newItem = items.get(newItemIndex); | |||
items = items.delete(newItemIndex); | |||
return newItem; | |||
}); | |||
list = list.concat(items); | |||
return list.sortBy(x => x.get('last_status'), (a, b) => compareId(a, b) * -1); | |||
}); | |||
} | |||
if (!next) { | |||
mutable.set('hasMore', false); | |||
} | |||
mutable.set('isLoading', false); | |||
}); | |||
}; | |||
export default function conversations(state = initialState, action) { | |||
switch (action.type) { | |||
case CONVERSATIONS_FETCH_REQUEST: | |||
return state.set('isLoading', true); | |||
case CONVERSATIONS_FETCH_FAIL: | |||
return state.set('isLoading', false); | |||
case CONVERSATIONS_FETCH_SUCCESS: | |||
return expandNormalizedConversations(state, action.conversations, action.next); | |||
case CONVERSATIONS_UPDATE: | |||
return updateConversation(state, action.conversation); | |||
default: | |||
return state; | |||
} | |||
}; |
@ -0,0 +1,111 @@ | |||
# frozen_string_literal: true | |||
# == Schema Information | |||
# | |||
# Table name: account_conversations | |||
# | |||
# id :bigint(8) not null, primary key | |||
# account_id :bigint(8) | |||
# conversation_id :bigint(8) | |||
# participant_account_ids :bigint(8) default([]), not null, is an Array | |||
# status_ids :bigint(8) default([]), not null, is an Array | |||
# last_status_id :bigint(8) | |||
# lock_version :integer default(0), not null | |||
# | |||
class AccountConversation < ApplicationRecord | |||
after_commit :push_to_streaming_api | |||
belongs_to :account | |||
belongs_to :conversation | |||
belongs_to :last_status, class_name: 'Status' | |||
before_validation :set_last_status | |||
def participant_account_ids=(arr) | |||
self[:participant_account_ids] = arr.sort | |||
end | |||
def participant_accounts | |||
if participant_account_ids.empty? | |||
[account] | |||
else | |||
Account.where(id: participant_account_ids) | |||
end | |||
end | |||
class << self | |||
def paginate_by_id(limit, options = {}) | |||
if options[:min_id] | |||
paginate_by_min_id(limit, options[:min_id]).reverse | |||
else | |||
paginate_by_max_id(limit, options[:max_id], options[:since_id]) | |||
end | |||
end | |||
def paginate_by_min_id(limit, min_id = nil) | |||
query = order(arel_table[:last_status_id].asc).limit(limit) | |||
query = query.where(arel_table[:last_status_id].gt(min_id)) if min_id.present? | |||
query | |||
end | |||
def paginate_by_max_id(limit, max_id = nil, since_id = nil) | |||
query = order(arel_table[:last_status_id].desc).limit(limit) | |||
query = query.where(arel_table[:last_status_id].lt(max_id)) if max_id.present? | |||
query = query.where(arel_table[:last_status_id].gt(since_id)) if since_id.present? | |||
query | |||
end | |||
def add_status(recipient, status) | |||
conversation = find_or_initialize_by(account: recipient, conversation_id: status.conversation_id, participant_account_ids: participants_from_status(recipient, status)) | |||
conversation.status_ids << status.id | |||
conversation.save | |||
conversation | |||
rescue ActiveRecord::StaleObjectError | |||
retry | |||
end | |||
def remove_status(recipient, status) | |||
conversation = find_by(account: recipient, conversation_id: status.conversation_id, participant_account_ids: participants_from_status(recipient, status)) | |||
return if conversation.nil? | |||
conversation.status_ids.delete(status.id) | |||
if conversation.status_ids.empty? | |||
conversation.destroy | |||
else | |||
conversation.save | |||
end | |||
conversation | |||
rescue ActiveRecord::StaleObjectError | |||
retry | |||
end | |||
private | |||
def participants_from_status(recipient, status) | |||
((status.mentions.pluck(:account_id) + [status.account_id]).uniq - [recipient.id]).sort | |||
end | |||
end | |||
private | |||
def set_last_status | |||
self.status_ids = status_ids.sort | |||
self.last_status_id = status_ids.last | |||
end | |||
def push_to_streaming_api | |||
return if destroyed? || !subscribed_to_timeline? | |||
PushConversationWorker.perform_async(id) | |||
end | |||
def subscribed_to_timeline? | |||
Redis.current.exists("subscribed:#{streaming_channel}") | |||
end | |||
def streaming_channel | |||
"timeline:direct:#{account_id}" | |||
end | |||
end |
@ -0,0 +1,7 @@ | |||
# frozen_string_literal: true | |||
class REST::ConversationSerializer < ActiveModel::Serializer | |||
attribute :id | |||
has_many :participant_accounts, key: :accounts, serializer: REST::AccountSerializer | |||
has_one :last_status, serializer: REST::StatusSerializer | |||
end |
@ -0,0 +1,12 @@ | |||
# frozen_string_literal: true | |||
class MuteWorker | |||
include Sidekiq::Worker | |||
def perform(account_id, target_account_id) | |||
FeedManager.instance.clear_from_timeline( | |||
Account.find(account_id), | |||
Account.find(target_account_id) | |||
) | |||
end | |||
end |
@ -0,0 +1,15 @@ | |||
# frozen_string_literal: true | |||
class PushConversationWorker | |||
include Sidekiq::Worker | |||
def perform(conversation_account_id) | |||
conversation = AccountConversation.find(conversation_account_id) | |||
message = InlineRenderer.render(conversation, conversation.account, :conversation) | |||
timeline_id = "timeline:direct:#{conversation.account_id}" | |||
Redis.current.publish(timeline_id, Oj.dump(event: :conversation, payload: message, queued_at: (Time.now.to_f * 1000.0).to_i)) | |||
rescue ActiveRecord::RecordNotFound | |||
true | |||
end | |||
end |
@ -0,0 +1,14 @@ | |||
class CreateAccountConversations < ActiveRecord::Migration[5.2] | |||
def change | |||
create_table :account_conversations do |t| | |||
t.belongs_to :account, foreign_key: { on_delete: :cascade } | |||
t.belongs_to :conversation, foreign_key: { on_delete: :cascade } | |||
t.bigint :participant_account_ids, array: true, null: false, default: [] | |||
t.bigint :status_ids, array: true, null: false, default: [] | |||
t.bigint :last_status_id, null: true, default: nil | |||
t.integer :lock_version, null: false, default: 0 | |||
end | |||
add_index :account_conversations, [:account_id, :conversation_id, :participant_account_ids], unique: true, name: 'index_unique_conversations' | |||
end | |||
end |
@ -0,0 +1,37 @@ | |||
require 'rails_helper' | |||
RSpec.describe Api::V1::ConversationsController, type: :controller do | |||
render_views | |||
let!(:user) { Fabricate(:user, account: Fabricate(:account, username: 'alice')) } | |||
let(:token) { Fabricate(:accessible_access_token, resource_owner_id: user.id, scopes: scopes) } | |||
let(:other) { Fabricate(:user, account: Fabricate(:account, username: 'bob')) } | |||
before do | |||
allow(controller).to receive(:doorkeeper_token) { token } | |||
end | |||
describe 'GET #index' do | |||
let(:scopes) { 'read:statuses' } | |||
before do | |||
PostStatusService.new.call(other.account, 'Hey @alice', nil, visibility: 'direct') | |||
end | |||
it 'returns http success' do | |||
get :index | |||
expect(response).to have_http_status(200) | |||
end | |||
it 'returns pagination headers' do | |||
get :index, params: { limit: 1 } | |||
expect(response.headers['Link'].links.size).to eq(2) | |||
end | |||
it 'returns conversations' do | |||
get :index | |||
json = body_as_json | |||
expect(json.size).to eq 1 | |||
end | |||
end | |||
end |
@ -0,0 +1,6 @@ | |||
Fabricator(:conversation_account) do | |||
account nil | |||
conversation nil | |||
participant_account_ids "" | |||
last_status nil | |||
end |
@ -0,0 +1,72 @@ | |||
require 'rails_helper' | |||
RSpec.describe AccountConversation, type: :model do | |||
let!(:alice) { Fabricate(:account, username: 'alice') } | |||
let!(:bob) { Fabricate(:account, username: 'bob') } | |||
let!(:mark) { Fabricate(:account, username: 'mark') } | |||
describe '.add_status' do | |||
it 'creates new record when no others exist' do | |||
status = Fabricate(:status, account: alice, visibility: :direct) | |||
status.mentions.create(account: bob) | |||
conversation = AccountConversation.add_status(alice, status) | |||
expect(conversation.participant_accounts).to include(bob) | |||
expect(conversation.last_status).to eq status | |||
expect(conversation.status_ids).to eq [status.id] | |||
end | |||
it 'appends to old record when there is a match' do | |||
last_status = Fabricate(:status, account: alice, visibility: :direct) | |||
conversation = AccountConversation.create!(account: alice, conversation: last_status.conversation, participant_account_ids: [bob.id], status_ids: [last_status.id]) | |||
status = Fabricate(:status, account: bob, visibility: :direct, thread: last_status) | |||
status.mentions.create(account: alice) | |||
new_conversation = AccountConversation.add_status(alice, status) | |||
expect(new_conversation.id).to eq conversation.id | |||
expect(new_conversation.participant_accounts).to include(bob) | |||
expect(new_conversation.last_status).to eq status | |||
expect(new_conversation.status_ids).to eq [last_status.id, status.id] | |||
end | |||
it 'creates new record when new participants are added' do | |||
last_status = Fabricate(:status, account: alice, visibility: :direct) | |||
conversation = AccountConversation.create!(account: alice, conversation: last_status.conversation, participant_account_ids: [bob.id], status_ids: [last_status.id]) | |||
status = Fabricate(:status, account: bob, visibility: :direct, thread: last_status) | |||
status.mentions.create(account: alice) | |||
status.mentions.create(account: mark) | |||
new_conversation = AccountConversation.add_status(alice, status) | |||
expect(new_conversation.id).to_not eq conversation.id | |||
expect(new_conversation.participant_accounts).to include(bob, mark) | |||
expect(new_conversation.last_status).to eq status | |||
expect(new_conversation.status_ids).to eq [status.id] | |||
end | |||
end | |||
describe '.remove_status' do | |||
it 'updates last status to a previous value' do | |||
last_status = Fabricate(:status, account: alice, visibility: :direct) | |||
status = Fabricate(:status, account: alice, visibility: :direct) | |||
conversation = AccountConversation.create!(account: alice, conversation: last_status.conversation, participant_account_ids: [bob.id], status_ids: [status.id, last_status.id]) | |||
last_status.mentions.create(account: bob) | |||
last_status.destroy! | |||
conversation.reload | |||
expect(conversation.last_status).to eq status | |||
expect(conversation.status_ids).to eq [status.id] | |||
end | |||
it 'removes the record if no other statuses are referenced' do | |||
last_status = Fabricate(:status, account: alice, visibility: :direct) | |||
conversation = AccountConversation.create!(account: alice, conversation: last_status.conversation, participant_account_ids: [bob.id], status_ids: [last_status.id]) | |||
last_status.mentions.create(account: bob) | |||
last_status.destroy! | |||
expect(AccountConversation.where(id: conversation.id).count).to eq 0 | |||
end | |||
end | |||
end |